feat: add telegram integration

.gitignore

@@ -4,4 +4,6 @@ __pycache__/
 
 .DS_Store
 .envrc
-.coverage
+.coverage
+
+.env

README.md

@@ -30,3 +30,18 @@ Event types are configured via environment variables:
 
 - `LogEvent`
   - `LOG_EVENT_LEVEL` - Level to log messages at
+
+- `TelegramEvent`
+  - `TELEGRAM_BOT_TOKEN` - API token for the Telegram bot
+
+## Finding the Telegram Group Chat ID
+
+To integrate Telegram events with the Observer, you need the Telegram group chat ID. Here's how you can find it:
+
+1. Open [Telegram Web](https://web.telegram.org).
+2. Navigate to the group chat for which you need the ID.
+3. Look at the URL in the browser's address bar; it should look something like `https://web.telegram.org/a/#-1111111111`.
+4. The group chat ID is the number in the URL, including the `-` sign if present (e.g., `-1111111111`).
+
+Use this ID in the `publishers.yaml` configuration to correctly set up Telegram events.
+

pyproject.toml

@@ -26,6 +26,7 @@ pyyaml = "^6.0"
 throttler = "1.2.1"
 types-pyyaml = "^6.0.12"
 types-pytz = "^2022.4.0.0"
+python-dotenv = "^1.0.1"
 
 
 [tool.poetry.group.dev.dependencies]

pyth_observer/__init__.py

@@ -22,6 +22,7 @@
 from pyth_observer.crosschain import CrosschainPrice
 from pyth_observer.crosschain import CrosschainPriceObserver as Crosschain
 from pyth_observer.dispatch import Dispatch
+from pyth_observer.models import Publisher
 
 PYTHTEST_HTTP_ENDPOINT = "https://api.pythtest.pyth.network/"
 PYTHTEST_WS_ENDPOINT = "wss://api.pythtest.pyth.network/"
@@ -49,7 +50,7 @@ class Observer:
     def __init__(
         self,
         config: Dict[str, Any],
-        publishers: Dict[str, str],
+        publishers: Dict[str, Publisher],
         coingecko_mapping: Dict[str, Symbol],
     ):
         self.config = config
@@ -134,8 +135,9 @@ async def run(self):
                     )
 
                     for component in price_account.price_components:
+                        pub = self.publishers.get(component.publisher_key.key, None)
                         publisher_name = (
-                            self.publishers.get(component.publisher_key.key, "")
+                            (pub.name if pub else "")
                             + f" ({component.publisher_key.key})"
                         ).strip()
                         states.append(

pyth_observer/cli.py

@@ -7,7 +7,8 @@
 from loguru import logger
 from prometheus_client import start_http_server
 
-from pyth_observer import Observer
+from pyth_observer import Observer, Publisher
+from pyth_observer.models import ContactInfo
 
 
 @click.command()
@@ -37,10 +38,26 @@
 )
 def run(config, publishers, coingecko_mapping, prometheus_port):
     config_ = yaml.safe_load(open(config, "r"))
-    publishers_ = yaml.safe_load(open(publishers, "r"))
-    publishers_inverted = {v: k for k, v in publishers_.items()}
+    # Load publishers YAML file and convert to dictionary of Publisher instances
+    publishers_raw = yaml.safe_load(open(publishers, "r"))
+    publishers_ = {
+        publisher["key"]: Publisher(
+            key=publisher["key"],
+            name=publisher["name"],
+            contact_info=(
+                ContactInfo(**publisher["contact_info"])
+                if "contact_info" in publisher
+                else None
+            ),
+        )
+        for publisher in publishers_raw
+    }
     coingecko_mapping_ = yaml.safe_load(open(coingecko_mapping, "r"))
-    observer = Observer(config_, publishers_inverted, coingecko_mapping_)
+    observer = Observer(
+        config_,
+        publishers_,
+        coingecko_mapping_,
+    )
 
     start_http_server(int(prometheus_port))
 

pyth_observer/dispatch.py

@@ -9,10 +9,12 @@
 from pyth_observer.check.publisher import PUBLISHER_CHECKS, PublisherState
 from pyth_observer.event import DatadogEvent  # Used dynamically
 from pyth_observer.event import LogEvent  # Used dynamically
+from pyth_observer.event import TelegramEvent  # Used dynamically
 from pyth_observer.event import Event
 
 assert DatadogEvent
 assert LogEvent
+assert TelegramEvent
 
 
 class Dispatch:

pyth_observer/event.py

@@ -1,20 +1,25 @@
 import os
 from typing import Dict, Literal, Protocol, TypedDict, cast
 
+import aiohttp
 from datadog_api_client.api_client import AsyncApiClient as DatadogAPI
 from datadog_api_client.configuration import Configuration as DatadogConfig
 from datadog_api_client.v1.api.events_api import EventsApi as DatadogEventAPI
 from datadog_api_client.v1.model.event_alert_type import EventAlertType
 from datadog_api_client.v1.model.event_create_request import EventCreateRequest
+from dotenv import load_dotenv
 from loguru import logger
 
 from pyth_observer.check import Check
 from pyth_observer.check.publisher import PublisherCheck
+from pyth_observer.models import Publisher
+
+load_dotenv()
 
 
 class Context(TypedDict):
     network: str
-    publishers: Dict[str, str]
+    publishers: Dict[str, Publisher]
 
 
 class Event(Protocol):
@@ -94,3 +99,47 @@ async def send(self):
 
         level = cast(LogEventLevel, os.environ.get("LOG_EVENT_LEVEL", "INFO"))
         logger.log(level, text.replace("\n", ". "))
+
+
+class TelegramEvent(Event):
+    def __init__(self, check: Check, context: Context):
+        self.check = check
+        self.context = context
+        self.telegram_bot_token = os.environ["TELEGRAM_BOT_TOKEN"]
+
+    async def send(self):
+        if self.check.__class__.__bases__ == (PublisherCheck,):
+            text = self.check.error_message()
+            publisher_key = self.check.state().public_key.key
+            publisher = self.context["publishers"].get(publisher_key, None)
+            # Ensure publisher is not None and has contact_info before accessing telegram_chat_id
+            chat_id = (
+                publisher.contact_info.telegram_chat_id
+                if publisher is not None and publisher.contact_info is not None
+                else None
+            )
+
+            if chat_id is None:
+                logger.warning(
+                    f"Telegram chat ID not found for publisher key {publisher_key}"
+                )
+                return
+
+            telegram_api_url = (
+                f"https://api.telegram.org/bot{self.telegram_bot_token}/sendMessage"
+            )
+            message_data = {
+                "chat_id": chat_id,
+                "text": text,
+                "parse_mode": "Markdown",
+            }
+
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    telegram_api_url, json=message_data
+                ) as response:
+                    if response.status != 200:
+                        response_text = await response.text()
+                        logger.error(
+                            f"Failed to send Telegram message: {response_text}"
+                        )

pyth_observer/models.py

@@ -0,0 +1,16 @@
+import dataclasses
+from typing import Optional
+
+
+@dataclasses.dataclass
+class ContactInfo:
+    telegram_chat_id: Optional[str] = None
+    email: Optional[str] = None
+    slack_channel_id: Optional[str] = None
+
+
+@dataclasses.dataclass
+class Publisher:
+    key: str
+    name: str
+    contact_info: Optional[ContactInfo] = None

sample.config.yaml

@@ -10,6 +10,7 @@ events:
   # NOTE: Uncomment to enable Datadog metrics, see README.md for datadog credential docs.
   # - DatadogEvent
   - LogEvent
+  - TelegramEvent
 checks:
   global:
     # Price feed checks

sample.publishers.yaml

@@ -1,4 +1,15 @@
-{
-  "publisher1": "66wJmrBqyykL7m4Erj4Ud29qhsm32DHSTo23zooupJrJ",
-  "publisher2": "3BkoB5MBSrrnDY7qe694UAuPpeMg7zJnodwbCnayNYzC",
-}
+- name: publisher1
+  key: "FR19oB2ePko2haah8yP4fhTycxitxkVQTxk3tssxX1Ce"
+  contact_info:
+    # Optional fields for contact information
+    telegram_chat_id: -4224704640
+    email:
+    slack_channel_id:
+
+- name: publisher2
+  key: "DgAK7fPveidN72LCwCF4QjFcYHchBZbtZnjEAtgU1bMX"
+  contact_info:
+    # Optional fields for contact information
+    telegram_chat_id: -4224704640
+    email:
+    slack_channel_id: