""" BSN Voice Chat — Browser-based voice dialog with DeepSeek v4 Flash. Architecture: Browser (Web Speech API + Audio) → Flask proxy → DeepSeek API TTS via Piper (switchable DE/EN/ES voices) Features: - Language selector: Deutsch | English | Español - DeepSeek v4 Flash as model (direct API, not OpenWebUI) - Stop button to interrupt speech - Thinking animation while AI processes - Light/Dark mode toggle - Mobile-responsive layout - Design with #20228a primary color """ import os import subprocess import hashlib import secrets import time import json from pathlib import Path from flask import Flask, request, jsonify, send_file, session, Response, stream_with_context from flask_cors import CORS from datetime import timedelta app = Flask(__name__) app.secret_key = secrets.token_hex(32) app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(hours=24) CORS(app, supports_credentials=True) # ── Config ────────────────────────────────────────────────────────────────── DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions" # Try environment first, then Hermes .env, then fallback _hermes_env = Path.home() / ".hermes" / ".env" if not os.environ.get("DEEPSEEK_API_KEY") and _hermes_env.exists(): with open(_hermes_env) as _f: for _line in _f: _line = _line.strip() if _line.startswith("DEEPSEEK_API_KEY="): os.environ["DEEPSEEK_API_KEY"] = _line.split("=", 1)[1] break DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "") DEEPSEEK_MODEL = "deepseek-v4-flash" APP_PASSWORD = os.environ.get("APP_PASSWORD", "hermes2026") PIPER_BIN = "/home/hermes/.local/bin/piper" PIPER_ENV = { "LD_LIBRARY_PATH": "/home/hermes/.local/lib", "ESPEAK_DATA_PATH": "/home/hermes/.local/share/espeak-ng-data", "PATH": os.environ.get("PATH", "/usr/bin"), } TTS_OUTPUT_DIR = Path("/tmp/voice-webapp-tts") TTS_OUTPUT_DIR.mkdir(parents=True, exist_ok=True) # ── Language Configuration ────────────────────────────────────────────────── LANGUAGES = { "de": { "label": "Deutsch", "flag": "🇩🇪", "stt_lang": "de-DE", "system_prompt": ( "Du bist ein hilfreicher Sprachassistent für die Brettspiel-Nachrichten-Website " "brettspiel-news.de. Antworte auf Deutsch. Keine Emojis. Kurze, natürliche Sätze " "(< 500 Zeichen). Du sprichst via Text-to-Speech, also verwende natürliche " "gesprochene Sprache." ), "piper_model": "/home/hermes/.hermes/cache/piper-voices/de_DE-thorsten-medium.onnx", }, "en": { "label": "English", "flag": "🇬🇧", "stt_lang": "en-US", "system_prompt": ( "You are a helpful voice assistant for a board game news website called " "brettspiel-news.de. Answer in English. No emojis. Short natural sentences " "(< 500 characters). You are speaking via text-to-speech, so use natural " "spoken language." ), "piper_model": "/home/hermes/.hermes/cache/piper-voices/en_US-joe-medium.onnx", }, "es": { "label": "Español", "flag": "🇪🇸", "stt_lang": "es-ES", "system_prompt": ( "Eres un asistente de voz útil para un sitio web de noticias de juegos de mesa " "llamado brettspiel-news.de. Responde en español. Sin emojis. Frases cortas y " "naturales (< 500 caracteres). Estás hablando mediante texto a voz, así que " "usa lenguaje hablado natural." ), "piper_model": "/home/hermes/.hermes/cache/piper-voices/es_ES-sharvard-medium.onnx", }, } # ── SPIEL 2025 Hallplan (OPTIONAL — set False to remove) ────────────────────── USE_HALLPLAN = True HALLPLAN_FILE = Path(__file__).parent / "data" / "spiel2025_aussteller.json" # Load hallplan data (empty list if disabled or missing) hallplan_data = [] if USE_HALLPLAN and HALLPLAN_FILE.exists(): try: with open(HALLPLAN_FILE) as f: hallplan_data = json.load(f) print(f"[HALLPLAN] Loaded {len(hallplan_data)} exhibitors from SPIEL 2025", flush=True) except Exception as e: print(f"[HALLPLAN] Error loading: {e}", flush=True) DEFAULT_LANG = "de" def get_lang(): """Return the current session language code (de|en|es).""" lang = session.get("language", DEFAULT_LANG) if lang not in LANGUAGES: lang = DEFAULT_LANG return lang # ── Auth ──────────────────────────────────────────────────────────────────── def check_auth(): return session.get("authenticated", False) # ── Routes ────────────────────────────────────────────────────────────────── @app.route("/") def index(): if not check_auth(): return LOGIN_PAGE return CHAT_PAGE @app.route("/login", methods=["POST"]) def login(): data = request.get_json(silent=True) or {} pw = data.get("password", "") if pw == APP_PASSWORD: session.permanent = True session["authenticated"] = True return jsonify({"ok": True}) time.sleep(0.8) return jsonify({"ok": False, "error": "Wrong password"}), 401 @app.route("/logout", methods=["POST"]) def logout(): session.clear() return jsonify({"ok": True}) @app.route("/api/language", methods=["GET", "POST"]) def api_language(): if not check_auth(): return jsonify({"error": "Not authenticated"}), 401 if request.method == "POST": data = request.get_json(silent=True) or {} lang = data.get("language", DEFAULT_LANG) if lang not in LANGUAGES: return jsonify({"error": f"Unsupported language: {lang}"}), 400 session["language"] = lang return jsonify({"ok": True, "language": lang}) # GET — return current language info lang = get_lang() return jsonify({ "ok": True, "language": lang, "languages": {k: {"label": v["label"], "flag": v["flag"], "stt_lang": v["stt_lang"]} for k, v in LANGUAGES.items()}, }) @app.route("/api/chat", methods=["POST"]) def chat(): if not check_auth(): return jsonify({"error": "Not authenticated"}), 401 data = request.get_json(silent=True) or {} user_message = data.get("message", "").strip() # Single source of truth: session language lang = get_lang() if not user_message: return jsonify({"error": "No message provided"}), 400 config = LANGUAGES[lang] # Optional: inject hallplan into system prompt for fair-related queries system_prompt = config["system_prompt"] # Hallplan data always injected when enabled if hallplan_data: hall_summary = "\n\nSPIEL 2025 Halle 1 — Aussteller und Stände (Referenzdaten):\n" for ex in hallplan_data: hall_summary += f"- {ex['name']}: Stand {ex['stand']}\n" system_prompt += hall_summary + "\nWenn jemand nach einem Aussteller, Stand, Halle, Messe oder SPIEL fragt, nutze diese Liste für die Antwort." import urllib.request payload = json.dumps({ "model": DEEPSEEK_MODEL, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message}, ], "max_tokens": 500, "temperature": 0.7, "stream": False, }).encode("utf-8") req = urllib.request.Request( DEEPSEEK_API_URL, data=payload, headers={ "Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json", }, ) try: with urllib.request.urlopen(req, timeout=120) as resp: raw = resp.read().decode("utf-8") result = json.loads(raw) choices = result.get("choices", []) if not choices: print(f"[CHAT] No choices in response: {str(result)[:300]}", flush=True) return jsonify({"error": "No response from model"}), 502 msg = choices[0].get("message", {}) reply = msg.get("content", "") if not reply or not reply.strip(): print(f"[CHAT] Empty content. Finish reason: {choices[0].get('finish_reason')} Full: {str(result)[:300]}", flush=True) fallback = { "de": "Entschuldigung, ich konnte keine Antwort generieren. Bitte versuche es erneut.", "en": "Sorry, I could not generate a response. Please try again.", "es": "Lo siento, no pude generar una respuesta. Por favor, inténtalo de nuevo.", } return jsonify({"ok": True, "reply": fallback.get(lang, fallback["de"])}) print(f"[CHAT] Reply: {len(reply)} chars [lang={lang}]", flush=True) return jsonify({"ok": True, "reply": reply, "language": lang}) except urllib.error.HTTPError as e: err_body = e.read().decode("utf-8", errors="replace") return jsonify({"error": f"API error: {err_body[:200]}"}), 502 except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/tts", methods=["POST"]) def tts(): if not check_auth(): return jsonify({"error": "Not authenticated"}), 401 data = request.get_json(silent=True) or {} text = data.get("text", "").strip() # Single source of truth: session language lang = get_lang() if not text: return jsonify({"error": "No text provided"}), 400 piper_model = LANGUAGES[lang]["piper_model"] text_hash = hashlib.md5((lang + ":" + text).encode()).hexdigest()[:12] out_path = TTS_OUTPUT_DIR / f"tts_{lang}_{text_hash}.wav" # Check cache (cached per language + text) if out_path.exists() and out_path.stat().st_size > 1000: return send_file(str(out_path), mimetype="audio/wav") text = text[:2000] try: result = subprocess.run( [PIPER_BIN, "--model", piper_model, "--output_file", str(out_path)], input=text.encode("utf-8"), capture_output=True, timeout=30, env=PIPER_ENV, ) if result.returncode != 0: return jsonify({"error": f"TTS failed: {result.stderr.decode()[:200]}"}), 500 if out_path.exists() and out_path.stat().st_size > 1000: return send_file(str(out_path), mimetype="audio/wav") return jsonify({"error": "Generated file empty"}), 500 except subprocess.TimeoutExpired: return jsonify({"error": "TTS timeout"}), 500 except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/check") def api_check(): return jsonify({"authenticated": check_auth()}) # ── Login Page ────────────────────────────────────────────────────────────── LOGIN_PAGE = """ BSN Voice Chat — Login

🎙 BSN Voice Chat

Voice dialog with AI — speak naturally and get spoken responses.

Wrong password

brettspiel-news.de
""" # ── Chat Page ─────────────────────────────────────────────────────────────── CHAT_PAGE = """ BSN Voice Chat

🎙 BSN Chat

Ready
☀️
🌙
🎙
Start speaking or type a message
Thinking
""" # ── Main ──────────────────────────────────────────────────────────────────── if __name__ == "__main__": print("Starting BSN Voice Chat on port 5001 (DeepSeek v4 Flash)...") app.run(host="::", port=5001, debug=False)