Add Telegram group and topic notifications

jcastro · jcastro · commit f3c4db6a8eb3 · 2026-04-16T11:34:06.000+02:00
diff --git a/.env.example b/.env.example
@@ -36,3 +36,5 @@ STOPLIGA_MAX_RESPONSE_BYTES=2097152
 # Optional Telegram notifications.
 # STOPLIGA_TELEGRAM_BOT_TOKEN=123456:replace-me
 # STOPLIGA_TELEGRAM_CHAT_ID=123456789
+# STOPLIGA_TELEGRAM_GROUP_ID=-1001234567890
+# STOPLIGA_TELEGRAM_TOPIC_ID=42
diff --git a/README.md b/README.md
@@ -133,8 +133,18 @@ STOPLIGA_MAX_RESPONSE_BYTES=2097152
 # STOPLIGA_GOTIFY_VERIFY_TLS=true
 # STOPLIGA_TELEGRAM_BOT_TOKEN=123456:replace-me
 # STOPLIGA_TELEGRAM_CHAT_ID=123456789
+# STOPLIGA_TELEGRAM_GROUP_ID=-1001234567890
+# STOPLIGA_TELEGRAM_TOPIC_ID=42
 ```
 
+Telegram options:
+
+- `STOPLIGA_TELEGRAM_CHAT_ID`: send to a private chat or to any chat id you already use today
+- `STOPLIGA_TELEGRAM_GROUP_ID`: explicit target for a Telegram group or supergroup
+- `STOPLIGA_TELEGRAM_TOPIC_ID`: optional forum topic id inside that Telegram group
+- set either `STOPLIGA_TELEGRAM_CHAT_ID` or `STOPLIGA_TELEGRAM_GROUP_ID`, not both
+- if `STOPLIGA_TELEGRAM_TOPIC_ID` is set, StopLiga sends the message with Telegram `message_thread_id`
+
 ## Sync Cycle
 
 With `STOPLIGA_SYNC_INTERVAL_SECONDS=300`, StopLiga runs a full sync every 5 minutes:
diff --git a/config.example.toml b/config.example.toml
@@ -51,3 +51,5 @@ verify_tls = false
 # gotify_verify_tls = true
 # telegram_bot_token = "123456:replace-me"
 # telegram_chat_id = "123456789"
+# telegram_group_id = "-1001234567890"
+# telegram_topic_id = 42
diff --git a/src/stopliga/config.py b/src/stopliga/config.py
@@ -200,6 +200,8 @@ def load_config_file(path: Path | None) -> dict[str, Any]:
         "gotify_priority": notifications.get("gotify_priority"),
         "telegram_bot_token": notifications.get("telegram_bot_token"),
         "telegram_chat_id": notifications.get("telegram_chat_id"),
+        "telegram_group_id": notifications.get("telegram_group_id"),
+        "telegram_topic_id": notifications.get("telegram_topic_id"),
         "notification_timeout": notifications.get("timeout"),
         "notification_retries": notifications.get("retries"),
         "notification_verify_tls": notifications.get("verify_tls"),
@@ -260,6 +262,8 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--gotify-priority", type=int, default=None, help="Gotify priority")
     parser.add_argument("--telegram-bot-token", default=None, help="Telegram bot token")
     parser.add_argument("--telegram-chat-id", default=None, help="Telegram user/chat id")
+    parser.add_argument("--telegram-group-id", default=None, help="Telegram group or supergroup id")
+    parser.add_argument("--telegram-topic-id", type=int, default=None, help="Telegram forum topic id")
     parser.add_argument("--notification-timeout", type=float, default=None, help="Notification HTTP timeout in seconds")
     parser.add_argument("--notification-retries", type=int, default=None, help="Notification retry count")
 
@@ -447,6 +451,13 @@ def load_config(args: argparse.Namespace, environ: Mapping[str, str] | None = No
         telegram_chat_id=str(
             _first(args.telegram_chat_id, _env_value(env, "STOPLIGA_TELEGRAM_CHAT_ID"), file_cfg.get("telegram_chat_id"), DEFAULTS.telegram_chat_id)
         ) if _first(args.telegram_chat_id, _env_value(env, "STOPLIGA_TELEGRAM_CHAT_ID"), file_cfg.get("telegram_chat_id"), DEFAULTS.telegram_chat_id) is not None else None,
+        telegram_group_id=str(
+            _first(args.telegram_group_id, _env_value(env, "STOPLIGA_TELEGRAM_GROUP_ID"), file_cfg.get("telegram_group_id"), DEFAULTS.telegram_group_id)
+        ) if _first(args.telegram_group_id, _env_value(env, "STOPLIGA_TELEGRAM_GROUP_ID"), file_cfg.get("telegram_group_id"), DEFAULTS.telegram_group_id) is not None else None,
+        telegram_topic_id=_parse_int(
+            _first(args.telegram_topic_id, _env_value(env, "STOPLIGA_TELEGRAM_TOPIC_ID"), file_cfg.get("telegram_topic_id")),
+            field_name="telegram_topic_id",
+        ) if _first(args.telegram_topic_id, _env_value(env, "STOPLIGA_TELEGRAM_TOPIC_ID"), file_cfg.get("telegram_topic_id")) is not None else None,
         notification_timeout=_parse_float(
             _first(args.notification_timeout, _env_value(env, "STOPLIGA_NOTIFICATION_TIMEOUT"), file_cfg.get("notification_timeout"), DEFAULTS.notification_timeout),
             field_name="notification_timeout",
@@ -509,8 +520,17 @@ def validate_config(config: Config, *, validate_connection: bool) -> None:
         raise ConfigError("Automatic route creation requires both STOPLIGA_VPN_NAME and STOPLIGA_TARGETS")
     if bool(config.gotify_url) != bool(config.gotify_token):
         raise ConfigError("Gotify notifications require both STOPLIGA_GOTIFY_URL and STOPLIGA_GOTIFY_TOKEN")
-    if bool(config.telegram_bot_token) != bool(config.telegram_chat_id):
-        raise ConfigError("Telegram notifications require both STOPLIGA_TELEGRAM_BOT_TOKEN and STOPLIGA_TELEGRAM_CHAT_ID")
+    if config.telegram_chat_id and config.telegram_group_id:
+        raise ConfigError("Set either STOPLIGA_TELEGRAM_CHAT_ID or STOPLIGA_TELEGRAM_GROUP_ID, not both")
+    telegram_target = config.resolved_telegram_chat_id()
+    if bool(config.telegram_bot_token) != bool(telegram_target):
+        raise ConfigError(
+            "Telegram notifications require STOPLIGA_TELEGRAM_BOT_TOKEN and either STOPLIGA_TELEGRAM_CHAT_ID or STOPLIGA_TELEGRAM_GROUP_ID"
+        )
+    if config.telegram_topic_id is not None and not telegram_target:
+        raise ConfigError("STOPLIGA_TELEGRAM_TOPIC_ID requires STOPLIGA_TELEGRAM_GROUP_ID or STOPLIGA_TELEGRAM_CHAT_ID")
+    if config.telegram_topic_id is not None and config.telegram_topic_id <= 0:
+        raise ConfigError("STOPLIGA_TELEGRAM_TOPIC_ID must be > 0")
     if validate_connection:
         _validate_unifi_host(config.host or "")
     if config.gotify_url:
diff --git a/src/stopliga/models.py b/src/stopliga/models.py
@@ -52,6 +52,8 @@ class Config:
     gotify_priority: int = 5
     telegram_bot_token: str | None = None
     telegram_chat_id: str | None = None
+    telegram_group_id: str | None = None
+    telegram_topic_id: int | None = None
     notification_timeout: float = 10.0
     notification_retries: int = 2
     notification_verify_tls: bool = True
@@ -68,9 +70,12 @@ def has_local_api_access(self) -> bool:
     def has_notifications(self) -> bool:
         return bool(
             (self.gotify_url and self.gotify_token)
-            or (self.telegram_bot_token and self.telegram_chat_id)
+            or (self.telegram_bot_token and self.resolved_telegram_chat_id())
         )
 
+    def resolved_telegram_chat_id(self) -> str | None:
+        return self.telegram_group_id or self.telegram_chat_id
+
     def resolved_health_max_age(self) -> int:
         if self.health_max_age_seconds is not None and self.health_max_age_seconds > 0:
             return self.health_max_age_seconds
diff --git a/src/stopliga/notifier.py b/src/stopliga/notifier.py
@@ -165,17 +165,21 @@ def send_notifications(config: Config, result: SyncResult, previous_state: dict[
             failures["gotify"] = str(exc)
             log_event(logger, logging.ERROR, "notification_provider_failed", provider="gotify", error=exc)
 
-    if config.telegram_bot_token and config.telegram_chat_id:
+    telegram_target = config.resolved_telegram_chat_id()
+    if config.telegram_bot_token and telegram_target:
         try:
             telegram_url = f"https://api.telegram.org/bot{config.telegram_bot_token}/sendMessage"
             request_config = _telegram_request_config(config)
+            telegram_payload = {
+                "chat_id": telegram_target,
+                "text": message,
+                "disable_web_page_preview": True,
+            }
+            if config.telegram_topic_id is not None:
+                telegram_payload["message_thread_id"] = config.telegram_topic_id
             _post_json(
                 telegram_url,
-                {
-                    "chat_id": config.telegram_chat_id,
-                    "text": message,
-                    "disable_web_page_preview": True,
-                },
+                telegram_payload,
                 timeout=request_config.timeout,
                 retries=request_config.retries,
                 verify_tls=request_config.verify_tls,
diff --git a/tests/test_config.py b/tests/test_config.py
@@ -144,6 +144,50 @@ def test_partial_telegram_configuration_is_rejected(self) -> None:
                 },
             )
 
+    def test_telegram_group_and_chat_id_are_mutually_exclusive(self) -> None:
+        parser = build_parser()
+        args = parser.parse_args([])
+        with self.assertRaises(ConfigError):
+            load_config(
+                args,
+                {
+                    "UNIFI_HOST": "10.0.0.2",
+                    "UNIFI_API_KEY": "test-api-key",
+                    "STOPLIGA_TELEGRAM_BOT_TOKEN": "123456:test",
+                    "STOPLIGA_TELEGRAM_CHAT_ID": "1234",
+                    "STOPLIGA_TELEGRAM_GROUP_ID": "-1001234567890",
+                },
+            )
+
+    def test_telegram_topic_requires_chat_target(self) -> None:
+        parser = build_parser()
+        args = parser.parse_args([])
+        with self.assertRaises(ConfigError):
+            load_config(
+                args,
+                {
+                    "UNIFI_HOST": "10.0.0.2",
+                    "UNIFI_API_KEY": "test-api-key",
+                    "STOPLIGA_TELEGRAM_TOPIC_ID": "42",
+                },
+            )
+
+    def test_telegram_group_and_topic_load_successfully(self) -> None:
+        parser = build_parser()
+        args = parser.parse_args([])
+        config = load_config(
+            args,
+            {
+                "UNIFI_HOST": "10.0.0.2",
+                "UNIFI_API_KEY": "test-api-key",
+                "STOPLIGA_TELEGRAM_BOT_TOKEN": "123456:test",
+                "STOPLIGA_TELEGRAM_GROUP_ID": "-1001234567890",
+                "STOPLIGA_TELEGRAM_TOPIC_ID": "42",
+            },
+        )
+        self.assertEqual(config.telegram_group_id, "-1001234567890")
+        self.assertEqual(config.telegram_topic_id, 42)
+
     def test_empty_route_name_is_rejected(self) -> None:
         parser = build_parser()
         args = parser.parse_args([])
diff --git a/tests/test_integration.py b/tests/test_integration.py
@@ -1054,6 +1054,57 @@ def fake_post_json(
             self.assertIn("Block status: ACTIVE -> INACTIVE", state.telegram_messages[0]["text"])
             self.assertIn("Blocking: INACTIVE", state.telegram_messages[0]["text"])
 
+    def test_telegram_notification_can_target_group_topic(self) -> None:
+        state = FakeState(
+            status_payload={"isBlocked": False},
+            ip_lines=["192.0.2.10"],
+            route={
+                "_id": "route-1",
+                "name": "LaLiga",
+                "enabled": True,
+                "network_id": "vpn-network-1",
+                "target_devices": [{"client_mac": "aa:bb:cc:dd:ee:01", "type": "CLIENT"}],
+                "ip_addresses": [{"ip_or_subnet": "192.0.2.10", "ip_version": "IPv4"}],
+            },
+        )
+        with tempfile.TemporaryDirectory() as tmpdir, TestServer(state, https=True) as unifi, TestServer(state, https=False) as feed:
+            state_path = Path(tmpdir) / "state.json"
+            state_path.write_text(json.dumps({"last_is_blocked": True}), encoding="utf-8")
+            config = self.make_config(
+                state_dir=tmpdir,
+                port=int(unifi.base_url.rsplit(":", 1)[1]),
+                status_url=f"{feed.base_url}/feed/status.json",
+                ip_list_url=f"{feed.base_url}/feed/ip_list.txt",
+                telegram_bot_token=f"{feed.base_url}/telegram/bot-token".replace(f"{feed.base_url}/", ""),
+                telegram_group_id="-1009876543210",
+                telegram_topic_id=77,
+            )
+            import stopliga.notifier as notifier  # noqa: WPS433
+
+            original_post_json = notifier._post_json
+
+            def fake_post_json(
+                url: str,
+                payload: dict[str, Any],
+                *,
+                timeout: float,
+                retries: int,
+                verify_tls: bool,
+                ca_file: Any,
+            ) -> None:
+                state.telegram_messages.append({"url": url, **payload})
+
+            notifier._post_json = fake_post_json
+            try:
+                StopLigaService(config).run_once()
+            finally:
+                notifier._post_json = original_post_json
+
+            self.assertEqual(len(state.telegram_messages), 1)
+            self.assertEqual(state.telegram_messages[0]["chat_id"], "-1009876543210")
+            self.assertEqual(state.telegram_messages[0]["message_thread_id"], 77)
+            self.assertIn("Block status: ACTIVE -> INACTIVE", state.telegram_messages[0]["text"])
+
     def test_notification_error_redacts_telegram_token(self) -> None:
         safe = _safe_notification_url("https://api.telegram.org/bot123456:secret-token/sendMessage")
         self.assertEqual(safe, "https://api.telegram.org/bot***/sendMessage")
