46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
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
|