New version and tests

Author: nopp

.gitignore

@@ -1 +1,30 @@
-flaskAlert.log
+# Logs
+*.log
+
+# Python
+__pycache__/
+*.py[cod]
+*.pyo
+*.pyd
+.Python
+
+# Virtual envs
+.venv/
+venv/
+env/
+
+# pytest
+.pytest_cache/
+.coverage
+htmlcov/
+
+# Distribution
+*.egg-info/
+dist/
+build/
+
+# IDE
+.venv/
+.idea/
+.vscode/
+*.swp

README.md

@@ -1,66 +1,119 @@
-# Alertmanager webhook for Telegram (Python Version)
+# Alertmanager webhook for Telegram (Python version)
 
 ![Docker Image CI](https://github.com/nopp/alertmanager-webhook-telegram-python/workflows/Docker%20Image%20CI/badge.svg)
 ![Code scanning - action](https://github.com/nopp/alertmanager-webhook-telegram-python/workflows/Code%20scanning%20-%20action/badge.svg)
 
-GO Version (https://github.com/nopp/alertmanager-webhook-telegram-go)
-
-Python version 3
-
-## INSTALL
-
-* pip install -r requirements.txt
-
-Change on flaskAlert.py
-=======================
-* botToken
-* chatID
-
-If you'll use with authentication, change too
-
-* XXXUSERNAME
-* XXXPASSWORD
-
-Disabling authentication
-========================
-On flaskAlert.py change app.config['BASIC_AUTH_FORCE'] = True to app.config['BASIC_AUTH_FORCE'] = False
-
-Alertmanager configuration example
-==================================
-
-	receivers:
-	- name: 'telegram-webhook'
-	  webhook_configs:
-	  - url: http://ipFlaskAlert:9119/alert
-	    send_resolved: true
-	    http_config:
-	      basic_auth:
-		username: 'admin'
-		password: 'password'
-
-One way to get the chat ID
-==========================
-1) Access https://web.telegram.org/
-2) Click to specific chat to the left
-3) At the url, you can get the chat ID(Ex: https://web.telegram.org/#/im?p=g1234567, so the chatID is 1234567)
-
-Running
-=======
-* python flaskAlert.py
-
-Running on docker
-=================
-    git clone https://github.com/nopp/alertmanager-webhook-telegram.git
-    cd alertmanager-webhook-telegram/docker/
-    docker build -t alertmanager-webhook-telegram:1.0 .
-
-    docker run -d --name telegram-bot \
-    	-e "bottoken=telegramBotToken" \
-    	-e "chatid=telegramChatID" \
-    	-e "username=<username>" \
-    	-e "password=<password>" \
-    	-p 9119:9119 alertmanager-webhook-telegram:1.0
-
-Example to test
-===============
-	curl -XPOST --data '{"status":"resolved","groupLabels":{"alertname":"instance_down"},"commonAnnotations":{"description":"i-0d7188fkl90bac100 of job ec2-sp-node_exporter has been down for more than 2 minutes.","summary":"Instance i-0d7188fkl90bac100 down"},"alerts":[{"status":"resolved","labels":{"name":"olokinho01-prod","instance":"i-0d7188fkl90bac100","job":"ec2-sp-node_exporter","alertname":"instance_down","os":"linux","severity":"page"},"endsAt":"2019-07-01T16:16:19.376244942-03:00","generatorURL":"http://pmts.io:9090","startsAt":"2019-07-01T16:02:19.376245319-03:00","annotations":{"description":"i-0d7188fkl90bac100 of job ec2-sp-node_exporter has been down for more than 2 minutes.","summary":"Instance i-0d7188fkl90bac100 down"}}],"version":"4","receiver":"infra-alert","externalURL":"http://alm.io:9093","commonLabels":{"name":"olokinho01-prod","instance":"i-0d7188fkl90bac100","job":"ec2-sp-node_exporter","alertname":"instance_down","os":"linux","severity":"page"}}' http://username:password@flaskAlert:9119/alert
+GO version: https://github.com/nopp/alertmanager-webhook-telegram-go
+
+## Project structure
+
+```
+├── app/
+│   ├── __init__.py     # app factory (create_app)
+│   ├── config.py       # config from env vars
+│   ├── alerts.py       # alert message builder
+│   ├── telegram.py     # Telegram HTTP client
+│   └── routes.py       # Flask blueprint (/alert)
+├── wsgi.py             # gunicorn / dev entrypoint
+├── requirements.txt
+└── docker/
+    ├── Dockerfile
+    └── run.sh
+```
+
+## Requirements
+
+- Python 3.11+
+- pip packages: see `requirements.txt`
+
+## Configuration
+
+All config via environment variables — no hardcoded secrets.
+
+| Variable | Required | Default | Description |
+|---|---|---|---|
+| `TELEGRAM_BOT_TOKEN` | ✅ | — | Bot token from @BotFather |
+| `TELEGRAM_CHAT_ID` | ✅ | — | Target chat/group ID |
+| `BASIC_AUTH_USERNAME` | ❌ | `""` | HTTP basic auth username |
+| `BASIC_AUTH_PASSWORD` | ❌ | `""` | HTTP basic auth password |
+| `BASIC_AUTH_FORCE` | ❌ | `true` | Set `false` to disable auth |
+| `SECRET_KEY` | ❌ | random | Flask secret key |
+| `MAX_RETRIES` | ❌ | `3` | Telegram send retries |
+
+## Install & run locally
+
+```bash
+pip install -r requirements.txt
+
+export TELEGRAM_BOT_TOKEN=your_token
+export TELEGRAM_CHAT_ID=your_chat_id
+export BASIC_AUTH_USERNAME=admin
+export BASIC_AUTH_PASSWORD=secret
+
+# dev
+python wsgi.py
+
+# production
+gunicorn -w 4 -b 0.0.0.0:9119 wsgi:app
+```
+
+## Docker
+
+```bash
+cd docker/
+docker build -t alertmanager-webhook-telegram .
+
+docker run -d --name telegram-webhook \
+  -e TELEGRAM_BOT_TOKEN=your_token \
+  -e TELEGRAM_CHAT_ID=your_chat_id \
+  -e BASIC_AUTH_USERNAME=admin \
+  -e BASIC_AUTH_PASSWORD=secret \
+  -p 9119:9119 \
+  alertmanager-webhook-telegram
+```
+
+## Alertmanager config example
+
+```yaml
+receivers:
+  - name: telegram-webhook
+    webhook_configs:
+      - url: http://your-host:9119/alert
+        send_resolved: true
+        http_config:
+          basic_auth:
+            username: admin
+            password: secret
+```
+
+## Get your Telegram chat ID
+
+1. Go to https://web.telegram.org/
+2. Open the target chat
+3. Copy the numeric ID from the URL (e.g. `https://web.telegram.org/#/im?p=g1234567` → ID is `-1234567` for groups)
+
+## Test
+
+```bash
+curl -XPOST \
+  -u admin:secret \
+  -H 'Content-Type: application/json' \
+  --data '{
+    "status": "resolved",
+    "alerts": [{
+      "status": "resolved",
+      "labels": {
+        "alertname": "instance_down",
+        "instance": "i-0d7188fkl90bac100",
+        "name": "prod-server"
+      },
+      "annotations": {
+        "summary": "Instance i-0d7188fkl90bac100 down",
+        "description": "Host has been down for more than 2 minutes."
+      },
+      "endsAt": "2024-01-01T16:16:19Z",
+      "startsAt": "2024-01-01T16:02:19Z"
+    }]
+  }' \
+  http://localhost:9119/alert
+```

app/__init__.py

@@ -0,0 +1,40 @@
+import logging
+
+from flask import Flask
+from flask_basicauth import BasicAuth
+
+from .config import Config
+from .routes import bp
+from .telegram import TelegramClient
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s %(message)s",
+)
+
+
+def create_app(config: Config | None = None) -> Flask:
+    app = Flask(__name__)
+
+    cfg = config or Config()
+
+    if not cfg.TELEGRAM_BOT_TOKEN:
+        raise ValueError("TELEGRAM_BOT_TOKEN is required")
+    if not cfg.TELEGRAM_CHAT_ID:
+        raise ValueError("TELEGRAM_CHAT_ID is required")
+    app.secret_key = cfg.SECRET_KEY
+    app.config["BASIC_AUTH_FORCE"] = cfg.BASIC_AUTH_FORCE
+    app.config["BASIC_AUTH_USERNAME"] = cfg.BASIC_AUTH_USERNAME
+    app.config["BASIC_AUTH_PASSWORD"] = cfg.BASIC_AUTH_PASSWORD
+
+    BasicAuth(app)
+
+    app.telegram = TelegramClient(  # type: ignore[attr-defined]
+        bot_token=cfg.TELEGRAM_BOT_TOKEN,
+        chat_id=cfg.TELEGRAM_CHAT_ID,
+        max_retries=cfg.MAX_RETRIES,
+    )
+
+    app.register_blueprint(bp)
+
+    return app

app/alerts.py

@@ -0,0 +1,26 @@
+from dateutil import parser
+
+
+def build_message(alert: dict) -> str:
+    """Build a human-readable Telegram message from a single Alertmanager alert."""
+    lines = [f"*Status:* {alert['status'].upper()}"]
+
+    labels = alert.get("labels", {})
+    if "instance" in labels:
+        name_suffix = f" ({labels['name']})" if "name" in labels else ""
+        lines.append(f"*Instance:* {labels['instance']}{name_suffix}")
+
+    annotations = alert.get("annotations", {})
+    for key in ("info", "summary", "description"):
+        if key in annotations:
+            lines.append(f"*{key.capitalize()}:* {annotations[key]}")
+
+    status = alert["status"]
+    if status == "resolved":
+        date_str = parser.parse(alert["endsAt"]).strftime("%Y-%m-%d %H:%M:%S")
+        lines.append(f"*Resolved at:* {date_str}")
+    elif status == "firing":
+        date_str = parser.parse(alert["startsAt"]).strftime("%Y-%m-%d %H:%M:%S")
+        lines.append(f"*Started at:* {date_str}")
+
+    return "\n".join(lines)

app/config.py

@@ -0,0 +1,11 @@
+import os
+
+
+class Config:
+    SECRET_KEY: str | bytes = os.environ.get("SECRET_KEY", os.urandom(24))
+    BASIC_AUTH_FORCE: bool = os.environ.get("BASIC_AUTH_FORCE", "true").lower() == "true"
+    BASIC_AUTH_USERNAME: str = os.environ.get("BASIC_AUTH_USERNAME", "")
+    BASIC_AUTH_PASSWORD: str = os.environ.get("BASIC_AUTH_PASSWORD", "")
+    TELEGRAM_BOT_TOKEN: str = os.environ.get("TELEGRAM_BOT_TOKEN", "")
+    TELEGRAM_CHAT_ID: str = os.environ.get("TELEGRAM_CHAT_ID", "")
+    MAX_RETRIES: int = int(os.environ.get("MAX_RETRIES", "3"))

app/routes.py

@@ -0,0 +1,45 @@
+import json
+import logging
+
+from flask import Blueprint, current_app, request
+
+from .alerts import build_message
+
+logger = logging.getLogger(__name__)
+
+bp = Blueprint("alerts", __name__)
+
+
+@bp.route("/alert", methods=["POST"])
+def post_alertmanager():
+    try:
+        payload = json.loads(request.get_data())
+    except json.JSONDecodeError as exc:
+        logger.error("Invalid JSON payload: %s", exc)
+        return "Bad Request", 400
+
+    alerts = payload.get("alerts", [])
+    if not alerts:
+        return "No alerts", 200
+
+    telegram = current_app.telegram
+    errors = []
+
+    for alert in alerts:
+        alert_name = alert.get("labels", {}).get("alertname", "unknown")
+        try:
+            message = build_message(alert)
+            telegram.send(message)
+            logger.info("Sent alert: %s", alert_name)
+        except Exception as exc:
+            logger.exception("Failed to send alert '%s': %s", alert_name, exc)
+            errors.append(f"{alert_name}: {exc}")
+
+    if errors:
+        try:
+            telegram.send(f"⚠️ Failed to forward {len(errors)} alert(s):\n" + "\n".join(errors))
+        except Exception:
+            pass
+        return "Partial failure", 207
+
+    return "OK", 200

app/telegram.py

@@ -0,0 +1,38 @@
+import logging
+import time
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+class TelegramClient:
+    def __init__(self, bot_token: str, chat_id: str, max_retries: int = 3) -> None:
+        self.chat_id = chat_id
+        self.max_retries = max_retries
+        self._url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
+
+    def send(self, text: str) -> None:
+        """Send a message to Telegram with retry on transient errors."""
+        payload = {"chat_id": self.chat_id, "text": text, "parse_mode": "Markdown"}
+
+        for attempt in range(1, self.max_retries + 1):
+            try:
+                resp = requests.post(self._url, json=payload, timeout=10)
+                resp.raise_for_status()
+                return
+            except requests.exceptions.HTTPError:
+                if resp.status_code == 429:
+                    wait = int(resp.headers.get("Retry-After", 30))
+                    logger.warning("Rate limited, waiting %ds (attempt %d/%d)…", wait, attempt, self.max_retries)
+                    time.sleep(wait)
+                else:
+                    raise
+            except requests.exceptions.Timeout:
+                logger.warning("Timeout (attempt %d/%d), retrying in 60s…", attempt, self.max_retries)
+                time.sleep(60)
+            except requests.exceptions.ConnectionError as exc:
+                logger.warning("Network error (attempt %d/%d): %s", attempt, self.max_retries, exc)
+                time.sleep(60)
+
+        raise RuntimeError(f"Failed to reach Telegram after {self.max_retries} attempts")