Spaces:
Paused
Paused
| """ | |
| Alternative notification methods for HF Spaces | |
| Since Telegram doesn't work in HF Spaces due to network restrictions, | |
| we provide alternative notification methods. | |
| """ | |
| import os | |
| import json | |
| import time | |
| from typing import Dict, Any | |
| class HFNotifications: | |
| """Handle notifications in HF Spaces environment""" | |
| def __init__(self): | |
| self.is_hf_space = self._check_hf_space() | |
| self.notification_log = "logs/notifications.jsonl" | |
| # Ensure log directory exists | |
| os.makedirs("logs", exist_ok=True) | |
| def _check_hf_space(self) -> bool: | |
| """Check if running in HF Spaces""" | |
| hf_env_vars = ['SPACE_ID', 'HF_SPACE_ID', 'HF_SPACE_HOST'] | |
| return any(os.getenv(var) for var in hf_env_vars) or 'hf.space' in os.getenv('SPACE_HOST', '') | |
| def log_notification(self, message: str, notification_type: str = "general"): | |
| """Log notification to file (works in HF Spaces)""" | |
| if not self.is_hf_space: | |
| return | |
| notification_data = { | |
| "timestamp": time.time(), | |
| "datetime": time.strftime('%Y-%m-%d %H:%M:%S UTC', time.gmtime()), | |
| "type": notification_type, | |
| "message": message, | |
| "space_id": os.getenv('SPACE_ID', 'unknown'), | |
| "host": os.getenv('SPACE_HOST', 'unknown') | |
| } | |
| try: | |
| with open(self.notification_log, 'a') as f: | |
| json.dump(notification_data, f) | |
| f.write('\n') | |
| print(f"[NOTIFICATION] {notification_type.upper()}: {message[:50]}...") | |
| except Exception as e: | |
| print(f"[ERROR] Failed to log notification: {e}") | |
| def get_recent_notifications(self, limit: int = 10) -> list: | |
| """Get recent notifications from log""" | |
| try: | |
| if not os.path.exists(self.notification_log): | |
| return [] | |
| notifications = [] | |
| with open(self.notification_log, 'r') as f: | |
| for line in f: | |
| if line.strip(): | |
| notifications.append(json.loads(line.strip())) | |
| return notifications[-limit:] | |
| except Exception as e: | |
| print(f"[ERROR] Failed to read notifications: {e}") | |
| return [] | |
| def clear_old_notifications(self, keep_days: int = 7): | |
| """Clear notifications older than specified days""" | |
| try: | |
| if not os.path.exists(self.notification_log): | |
| return | |
| cutoff_time = time.time() - (keep_days * 24 * 60 * 60) | |
| recent_notifications = [] | |
| with open(self.notification_log, 'r') as f: | |
| for line in f: | |
| if line.strip(): | |
| notification = json.loads(line.strip()) | |
| if notification.get('timestamp', 0) > cutoff_time: | |
| recent_notifications.append(notification) | |
| with open(self.notification_log, 'w') as f: | |
| for notification in recent_notifications: | |
| json.dump(notification, f) | |
| f.write('\n') | |
| print(f"[INFO] Cleared old notifications, kept {len(recent_notifications)} recent ones") | |
| except Exception as e: | |
| print(f"[ERROR] Failed to clear notifications: {e}") | |
| # Global instance | |
| hf_notifications = HFNotifications() | |
| def send_hf_notification(message: str, notification_type: str = "general"): | |
| """Send notification in HF Spaces (logs to file)""" | |
| hf_notifications.log_notification(message, notification_type) | |
| def get_hf_notifications(limit: int = 10) -> list: | |
| """Get recent notifications from HF Spaces log""" | |
| return hf_notifications.get_recent_notifications(limit) | |
| # Example usage: | |
| if __name__ == "__main__": | |
| # Test notifications | |
| send_hf_notification("🚀 Bot started successfully", "system") | |
| send_hf_notification("💰 BTCUSDT LONG @ $84,200", "trade") | |
| send_hf_notification("📊 Daily P&L: $15.23", "performance") | |
| # Get recent notifications | |
| recent = get_hf_notifications(5) | |
| print(f"Recent notifications: {len(recent)}") | |
| for notif in recent[-3:]: | |
| print(f" {notif['datetime']}: {notif['message'][:50]}...") | |