import requests import os from datetime import datetime from dotenv import load_dotenv # Load environment variables from .env file load_dotenv() class TelegramNotifier: """ A class to send notifications to Telegram when parts with red status are detected. To use this class: 1. Create a Telegram bot using BotFather and get the API token 2. Get the chat ID where messages should be sent 3. Set environment variables TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID 4. Or pass these values directly to the constructor """ def __init__(self, bot_token=None, chat_id=None): """ Initialize the TelegramNotifier with bot token and chat ID. Args: bot_token (str, optional): Telegram Bot API token. If None, reads from TELEGRAM_BOT_TOKEN env var. chat_id (str, optional): Telegram chat ID to send messages to. If None, reads from TELEGRAM_CHAT_ID env var. """ # Use provided parameters first, fall back to environment variables if not provided self.bot_token = bot_token if bot_token is not None else os.environ.get('TELEGRAM_BOT_TOKEN') self.chat_id = chat_id if chat_id is not None else os.environ.get('TELEGRAM_CHAT_ID') if not self.bot_token: print("Warning: Telegram bot token not provided. Notifications will not be sent.") if not self.chat_id: print("Warning: Telegram chat ID not provided. Notifications will not be sent.") def send_message(self, message): """ Send a message to the Telegram chat. Args: message (str): The message to send Returns: bool: True if message was sent successfully, False otherwise """ if not self.bot_token or not self.chat_id: print("Cannot send Telegram message: bot token or chat ID is missing") return False url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage" payload = { "chat_id": self.chat_id, "text": message, "parse_mode": "HTML" } try: response = requests.post(url, data=payload) response.raise_for_status() return True except Exception as e: print(f"Error sending Telegram message: {e}") return False def check_and_notify(self, results, base_url="http://localhost:5000"): """ Check for parts with red status and send a notification if any are found. Args: results (list): List of dictionaries containing equipment and part information. Each dictionary should have 'equipment_id', 'equipment_name', 'label', 'latest_created_at', 'latest_value', 'messages', 'part_id', and 'status' keys. base_url (str, optional): Base URL for the link to more detailed data Returns: bool: True if notification was sent, False otherwise """ # Check if results is a list of dictionaries with 'parts' key (old format) # The old format had a nested structure where each equipment had a 'parts' key # containing a list of part dictionaries if results and isinstance(results, list) and 'parts' in results[0]: # Extract all parts from the nested structure (backward compatibility) all_parts = [] for equipment in results: if 'parts' in equipment: all_parts.extend(equipment['parts']) else: # Use the flat list directly (new format) # The new format is a flat list of dictionaries, each containing information about a part # with fields like equipment_id, equipment_name, label, latest_created_at, latest_value, # messages, part_id, and status all_parts = results # Find parts with different status colors red_parts = [part for part in all_parts if part['status'] == 'red'] yellow_parts = [part for part in all_parts if part['status'] == 'yellow'] green_parts = [part for part in all_parts if part['status'] == 'green'] if not red_parts: print("No parts with red status found. No notification needed.") return False # Create notification message with the required format message = "Telah terjadi anomali data di PI yang diidentifikasi, dengan contoh data sebagai berikut:\n\n" # Add only the top 3 parts with red status for i, part in enumerate(red_parts[:3], 1): part_label = part.get('label', f"Part ID: {part['part_id']}") message += f"{i}. {part_label}\n" # If there are more than 3 parts with red status, add an ellipsis if len(red_parts) > 3: message += "4. ...\n" # Add count of status colors as a list message += f"\nJumlah status:\n" message += f"- Merah: {len(red_parts)}\n" message += f"- Kuning: {len(yellow_parts)}\n" message += f"- Hijau: {len(green_parts)}\n" # Add link to more detailed data message += f"\nKlik disini untuk data lebih lengkap:\n" message += f"Klik Disini\n\n" # Add signature message += "Salam,\nAdmin Digital Twin" # Send the notification return self.send_message(message) def check_red_status_and_notify(results, bot_token=None, chat_id=None, base_url="http://localhost:5000"): """ Convenience function to check for red status parts and send a notification. Args: results (list): List of dictionaries containing equipment and part information. Each dictionary should have 'equipment_id', 'equipment_name', 'label', 'latest_created_at', 'latest_value', 'messages', 'part_id', and 'status' keys. Also supports the old format with nested 'parts' key for backward compatibility. bot_token (str, optional): Telegram Bot API token chat_id (str, optional): Telegram chat ID to send messages to base_url (str, optional): Base URL for the link to more detailed data Returns: bool: True if notification was sent, False otherwise """ notifier = TelegramNotifier(bot_token, chat_id) return notifier.check_and_notify(results, base_url) def kirimpesan(results, bot_token=None, chat_id=None, base_url="http://localhost:5000"): """ Convenience function to check for red status parts and send a notification. Args: results (list): List of dictionaries containing equipment and part information. Each dictionary should have 'equipment_id', 'equipment_name', 'label', 'latest_created_at', 'latest_value', 'messages', 'part_id', and 'status' keys. Also supports the old format with nested 'parts' key for backward compatibility. bot_token (str, optional): Telegram Bot API token chat_id (str, optional): Telegram chat ID to send messages to base_url (str, optional): Base URL for the link to more detailed data Returns: bool: True if notification was sent, False otherwise """ notifier = TelegramNotifier(bot_token, chat_id) return notifier.check_and_notify(results, base_url) if __name__ == "__main__": # Example usage with the new flat list data structure example_results = [ { "equipment_id": "0096ab37-14b0-4946-b7e3-76160f501f3e", "equipment_name": "MBFP MOTOR B", "label": "3FW-TE324 | MBFP MTR B BRG 1 TEMP", "latest_created_at": "Tue, 29 Jul 2025 20:00:00 GMT", "latest_value": 35.8, "messages": [], "part_id": "15ac1ec9-6681-48a8-a59b-a0ff8a681ddf", "status": "green" }, { "equipment_id": "0096ab37-14b0-4946-b7e3-76160f501f3e", "equipment_name": "MBFP MOTOR B", "label": "3FW-TE325 | MBFP MTR B BRG 2 TEMP", "latest_created_at": "Tue, 29 Jul 2025 20:00:00 GMT", "latest_value": 33.4, "messages": [], "part_id": "529a0f2e-65e2-4cb3-aedb-cbd0678fbcee", "status": "green" }, { "equipment_id": "0096ab37-14b0-4946-b7e3-76160f501f3e", "equipment_name": "MBFP MOTOR B", "label": "3FW-P300-A2 | MBFP MTR B CUR", "latest_created_at": "Tue, 29 Jul 2025 20:00:00 GMT", "latest_value": 0.0, "messages": [ "Error: Contains null or zero values", "Error: Same value (0.0) for 3 consecutive days" ], "part_id": "5be40d7a-ad0b-4e31-b6ab-d755004efef7", "status": "red" }, { "equipment_id": "0096ab37-14b0-4946-b7e3-76160f501f3e", "equipment_name": "MBFP MOTOR B", "label": "3FW-TE320 | MBFP MTR B WDG TEMP W", "latest_created_at": "Tue, 29 Jul 2025 20:00:00 GMT", "latest_value": 34.81954, "messages": [], "part_id": "9726b4c9-a7f0-4adf-b493-06bd2973dc62", "status": "green" }, { "equipment_id": "0096ab37-14b0-4946-b7e3-76160f501f3e", "equipment_name": "MBFP MOTOR B", "label": "3FW-TE316 | MBFP MTR B WDG TEMP U", "latest_created_at": "Tue, 29 Jul 2025 20:00:00 GMT", "latest_value": 35.4, "messages": [], "part_id": "a148a27d-1ef4-4301-9fbe-a22def4e491c", "status": "green" } ] # Test the notification with a sample base_url base_url = "http://example.com" check_red_status_and_notify(example_results, base_url=base_url)