92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
"""
|
|
Telegram notification service
|
|
"""
|
|
import logging
|
|
import time
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
from config import settings
|
|
|
|
try:
|
|
import requests
|
|
REQUESTS_AVAILABLE = True
|
|
except ImportError:
|
|
REQUESTS_AVAILABLE = False
|
|
|
|
|
|
class TelegramService:
|
|
"""Service for sending Telegram notifications"""
|
|
|
|
def __init__(self):
|
|
self.logger = logging.getLogger(__name__)
|
|
self._token: Optional[str] = None
|
|
self._chat_id: Optional[str] = None
|
|
self._last_error_cache: dict = {}
|
|
|
|
def configure(self, token: str, chat_id: str) -> None:
|
|
"""Configure Telegram credentials"""
|
|
self._token = token
|
|
self._chat_id = chat_id
|
|
self.logger.info("Telegram service configured")
|
|
|
|
@property
|
|
def is_configured(self) -> bool:
|
|
"""Check if Telegram is configured"""
|
|
return bool(self._token and self._chat_id)
|
|
|
|
def _send_request(self, endpoint: str, data: dict, retries: int = 3, delay: int = 2) -> bool:
|
|
"""Make API request to Telegram"""
|
|
if not REQUESTS_AVAILABLE:
|
|
self.logger.warning("requests library not available")
|
|
return False
|
|
|
|
url = f"https://api.telegram.org/bot{self._token}/{endpoint}"
|
|
|
|
for attempt in range(retries):
|
|
try:
|
|
resp = requests.post(url, data=data, timeout=10)
|
|
if resp.status_code == 200:
|
|
return True
|
|
else:
|
|
self.logger.error(f"Telegram API error: {resp.status_code}")
|
|
except Exception as e:
|
|
self.logger.error(f"Telegram request failed (attempt {attempt+1}/{retries}): {e}")
|
|
time.sleep(delay)
|
|
return False
|
|
|
|
def send_message(self, message: str) -> bool:
|
|
"""Send a text message to Telegram"""
|
|
if not self.is_configured:
|
|
self.logger.warning("Telegram not configured, skipping notification")
|
|
return False
|
|
data = {"chat_id": self._chat_id, "text": message}
|
|
return self._send_request("sendMessage", data)
|
|
|
|
def send_start_notification(self) -> bool:
|
|
"""Send service start notification"""
|
|
message = "CBCFacil Service Started - AI document processing active"
|
|
return self.send_message(message)
|
|
|
|
def send_error_notification(self, error_key: str, error_message: str) -> bool:
|
|
"""Send error notification with throttling"""
|
|
now = datetime.utcnow()
|
|
prev = self._last_error_cache.get(error_key)
|
|
if prev is None:
|
|
self._last_error_cache[error_key] = (error_message, now)
|
|
else:
|
|
prev_msg, prev_time = prev
|
|
if error_message != prev_msg or (now - prev_time).total_seconds() > settings.ERROR_THROTTLE_SECONDS:
|
|
self._last_error_cache[error_key] = (error_message, now)
|
|
else:
|
|
return False
|
|
return self.send_message(f"Error: {error_message}")
|
|
|
|
|
|
# Global instance
|
|
telegram_service = TelegramService()
|
|
|
|
|
|
def send_telegram_message(message: str, retries: int = 3, delay: int = 2) -> bool:
|
|
"""Legacy function for backward compatibility"""
|
|
return telegram_service.send_message(message)
|