ohmu/dev/config/alert-relay/alert_relay.py

#!/usr/bin/env python3
"""HyperDX alert-webhook → Telegram relay (ohmu).

Võtab vastu HyperDX "generic" webhook POST-i ja saadab sisu Telegrami.
Kasutab ainult standardteeki (ei vaja pip-pakette).

Env: TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID (vaikimisi 5036685258)
Kuulab: 0.0.0.0:9099
"""
from __future__ import annotations

import json
import os
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer

TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "").strip()
CHAT = os.environ.get("TELEGRAM_CHAT_ID", "5036685258")
PORT = int(os.environ.get("PORT", "9099"))


def send(text: str) -> None:
    if not TOKEN:
        print("[relay] TELEGRAM_BOT_TOKEN puudub", flush=True)
        return
    data = urllib.parse.urlencode({"chat_id": CHAT, "text": text[:4000], "parse_mode": "Markdown"}).encode()
    req = urllib.request.Request(f"https://api.telegram.org/bot{TOKEN}/sendMessage", data=data)
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            r.read()
    except Exception as e:  # noqa: BLE001
        print("[relay] Telegram viga:", str(e)[:200], flush=True)


def format_msg(data: dict) -> str:
    def pick(*keys):
        for k in keys:
            v = data.get(k) or (data.get("alert") or {}).get(k)
            if v not in (None, ""):
                return v
        return None

    name = pick("name", "alertName", "title") or "HyperDX alert"
    state = pick("state", "status", "event") or "?"
    icon = {"ALERT": "🔴", "alert": "🔴", "RESOLVED": "✅", "resolved": "✅", "OK": "✅"}.get(str(state), "⚠️")
    lines = [f"{icon} *{name}*", f"Seis: `{state}`"]
    msg = pick("message", "note", "description")
    if msg:
        lines.append(str(msg)[:800])
    if not msg:
        lines.append("```\n" + json.dumps(data, ensure_ascii=False)[:1200] + "\n```")
    return "\n".join(lines)


class H(BaseHTTPRequestHandler):
    def do_POST(self):  # noqa: N802
        n = int(self.headers.get("Content-Length", "0") or 0)
        raw = self.rfile.read(n) if n else b"{}"
        try:
            data = json.loads(raw or b"{}")
        except Exception:  # noqa: BLE001
            data = {"message": raw.decode("utf-8", "replace")[:1000]}
        send(format_msg(data if isinstance(data, dict) else {"message": str(data)}))
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(b'{"ok":true}')

    def do_GET(self):  # noqa: N802
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"alert-relay ok")

    def log_message(self, *args):  # vaikimisi logi maha
        return


if __name__ == "__main__":
    print(f"[relay] kuulan :{PORT} -> Telegram {CHAT}", flush=True)
    HTTPServer(("0.0.0.0", PORT), H).serve_forever()