from __future__ import annotations import hashlib import hmac import json import os from urllib import parse, request def send(status: str, filename: str = "", detail: str = "") -> bool: backend = os.getenv("ASR_NOTIFY_BACKEND", "none").strip().lower() try: if backend == "telegram": token = os.getenv("TELEGRAM_BOT_TOKEN", "") chat = os.getenv("TELEGRAM_CHAT_ID", "") if not token or not chat: return False body = parse.urlencode({ "chat_id": chat, "text": "\n".join(x for x in (status, filename, detail) if x), }).encode() req = request.Request( f"https://api.telegram.org/bot{token}/sendMessage", data=body, method="POST") elif backend == "hermes": url = os.getenv("HERMES_WEBHOOK_URL", "") secret = os.getenv("HERMES_WEBHOOK_SECRET", "") if not url or not secret: return False body = json.dumps( {"status": status, "filename": filename, "detail": detail}, ensure_ascii=False).encode() sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() req = request.Request( url, data=body, method="POST", headers={ "Content-Type": "application/json", "X-Webhook-Signature": sig, }) else: return False with request.urlopen(req, timeout=10) as response: return 200 <= response.status < 300 except Exception: return False