Coverage for app\services\notification_service.py: 16%

1import logging 

2import httpx 

3from typing import Optional, List, Dict, Any 

4from sqlalchemy.orm import Session 

5from ..db import SystemSetting, Notification, NotificationType 

6 

7logger = logging.getLogger(__name__) 

8 

9class NotificationService: 

10 def __init__(self, db: Session): 

11 self.db = db 

12 self.settings = self._load_settings() 

13 

14 def _load_settings(self) -> Dict[str, Any]: 

15 # Load all relevant settings into a dictionary 

16 keys = [ 

17 "onesignal_app_id", "onesignal_api_key", 

18 "resend_api_key", "resend_from_email", 

19 "telegram_bot_token" 

20 ] 

21 settings = {} 

22 for key in keys: 

23 setting = self.db.query(SystemSetting).filter(SystemSetting.key == key).first() 

24 if setting and setting.value: 

25 # Handle both direct values and dict wrapped values 

26 if isinstance(setting.value, dict): 

27 settings[key] = setting.value.get("value", setting.value) 

28 else: 

29 settings[key] = setting.value 

30 return settings 

31 

32 async def send_push(self, title: str, message: str, player_ids: Optional[List[str]] = None) -> bool: 

33 app_id = self.settings.get("onesignal_app_id") 

34 api_key = self.settings.get("onesignal_api_key") 

35 

36 if not app_id or not api_key: 

37 logger.warning("OneSignal settings missing. Skipping push.") 

38 return False 

39 

40 headers = { 

41 "Content-Type": "application/json; charset=utf-8", 

42 "Authorization": f"Basic {api_key}" 

43 } 

44 

45 payload = { 

46 "app_id": app_id, 

47 "headings": {"en": title}, 

48 "contents": {"en": message}, 

49 } 

50 

51 if player_ids: 

52 payload["include_player_ids"] = player_ids 

53 else: 

54 payload["included_segments"] = ["All"] 

55 

56 async with httpx.AsyncClient() as client: 

57 try: 

58 response = await client.post("https://onesignal.com/api/v1/notifications", json=payload, headers=headers) 

59 response.raise_for_status() 

60 logger.info(f"Push notification sent: {response.json()}") 

61 return True 

62 except Exception as e: 

63 logger.error(f"Failed to send push notification: {e}") 

64 return False 

65 

66 async def send_email(self, to_email: str, subject: str, html_content: str) -> bool: 

67 api_key = self.settings.get("resend_api_key") 

68 from_email = self.settings.get("resend_from_email") 

69 

70 if not api_key or not from_email: 

71 logger.warning("Resend settings missing. Skipping email.") 

72 return False 

73 

74 headers = { 

75 "Authorization": f"Bearer {api_key}", 

76 "Content-Type": "application/json" 

77 } 

78 

79 payload = { 

80 "from": from_email, 

81 "to": [to_email], 

82 "subject": subject, 

83 "html": html_content 

84 } 

85 

86 async with httpx.AsyncClient() as client: 

87 try: 

88 response = await client.post("https://api.resend.com/emails", json=payload, headers=headers) 

89 response.raise_for_status() 

90 logger.info(f"Email sent: {response.json()}") 

91 return True 

92 except Exception as e: 

93 logger.error(f"Failed to send email: {e}") 

94 return False 

95 

96 async def send_telegram(self, chat_id: str, message: str) -> bool: 

97 bot_token = self.settings.get("telegram_bot_token") 

98 

99 if not bot_token: 

100 logger.warning("Telegram bot token missing. Skipping message.") 

101 return False 

102 

103 url = f"https://api.telegram.org/bot{bot_token}/sendMessage" 

104 payload = { 

105 "chat_id": chat_id, 

106 "text": message, 

107 "parse_mode": "HTML" 

108 } 

109 

110 async with httpx.AsyncClient() as client: 

111 try: 

112 response = await client.post(url, json=payload) 

113 response.raise_for_status() 

114 logger.info(f"Telegram message sent: {response.json()}") 

115 return True 

116 except Exception as e: 

117 logger.error(f"Failed to send telegram message: {e}") 

118 return False 

119 

120 async def dispatch_notification(self, notification: Notification, user_email: Optional[str] = None, user_telegram_id: Optional[str] = None): 

121 """ 

122 Dispatches notification to all enabled channels based on availability. 

123 """ 

124 tasks = [] 

125 

126 # Always try push (broadcast or specific) 

127 # Note: For specific users, we need their OneSignal Player ID.  

128 # Assuming we might store it in user profile or handle it client-side. 

129 # For now, if broadcast, send to All. 

130 if not notification.recipient_id: 

131 tasks.append(self.send_push(notification.title, notification.message)) 

132 

133 # Email 

134 if user_email and self.settings.get("resend_api_key"): 

135 tasks.append(self.send_email(user_email, notification.title, f"<h1>{notification.title}</h1><p>{notification.message}</p>")) 

136 

137 # Telegram 

138 if user_telegram_id and self.settings.get("telegram_bot_token"): 

139 tasks.append(self.send_telegram(user_telegram_id, f"*{notification.title}*\n{notification.message}")) 

140 

141 # Execute all (pseudo-parallel if we used gather, but for now linear awaits or background task) 

142 for task in tasks: 

143 await task