import json import requests from django.db.models import Q from onesignal_sdk.client import Client as OneSignalClient from django.conf import settings from accounts.models import IAmPrincipal from .models import IAmPrincipalNotificationSettings, PrincipalType import logging logger = logging.getLogger(__name__) def onesignal_send_notification( title, message, image_url=None, eligible_principals=None ): onesignal_app_id = settings.ONE_SIGNAL_APP_ID onesignal_rest_api_key = settings.ONE_SIGNAL_API_KEY headers = { "Content-Type": "application/json; charset=utf-8", "Authorization": f"Basic {onesignal_rest_api_key}", } if not eligible_principals: return None # Or handle this scenario as needed # Extract OneSignal player IDs # player_ids = eligible_principals.values_list("player_id", flat=True) # Construct the notification payload data = { "app_id": onesignal_app_id, "headings": {"en": title}, "contents": {"en": message}, "include_player_ids": eligible_principals, # Include the filtered player IDs # Add other optional notification data according to OneSignal documentation } if image_url: # Include image URL if provided (requires additional OneSignal configuration) data["big_picture"] = str(image_url.url) data["small_icon"] = str(image_url.url) data["large_icon"] = str(image_url.url) # data = json.dumps(data) print("Data: ", data) # Send the notification request to OneSignal response = requests.post( "https://onesignal.com/api/v1/notifications", headers=headers, json=data ) if response.status_code == 200: print("Notification sent successfully!") return True # Indicate success else: print(f"Error sending notification: {response.text}") return False # Indicate failure def send_notification(title, message, image_url=None, eligible_principals=None): # Initialize OneSignal client onesignal_client = OneSignalClient( app_id=settings.ONE_SIGNAL_APP_ID, rest_api_key=settings.ONE_SIGNAL_API_KEY ) if not eligible_principals: return None # Or handle this scenario as needed # Extract OneSignal player IDs # player_ids = eligible_principals.values_list("player_id", flat=True) # Prepare notification payload notification_payload = { "headings": {"en": title}, "contents": {"en": message}, "include_player_ids": eligible_principals, } # if image_url: # # image_url_str = image_url.url # notification_payload["big_picture"] = image_url # notification_payload["small_icon"] = image_url # notification_payload["large_icon"] = image_url # Send notification try: response = onesignal_client.send_notification(notification_payload) logger.info(f"Notification sent successfully: {response.status_code} {response.body}") except Exception as e: logger.error(f"An unexpected error occurred: {str(e)}") return None return response # def get_eligible_principals_for_notification(push_notification): # principal_type = push_notification.principal_type # notification_category = push_notification.notification_category # if principal_type == PrincipalType.BOTH: # # If BOTH is selected, fetch users categorized under both EVENT_USER and EVENT_MANAGER # eligible_principals = list( # IAmPrincipal.objects.filter( # Q(principal_type__name=PrincipalType.EVENT_USER) # | Q(principal_type__name=PrincipalType.EVENT_MANAGER), # notifications_principal__notification_category=notification_category, # notifications_principal__is_enabled=True, # ).values_list("player_id", flat=True) # ) # elif principal_type == PrincipalType.EVENT_USER: # # Fetch only EVENT_USER principals # eligible_principals = list( # IAmPrincipal.objects.filter( # principal_type__name=PrincipalType.EVENT_USER, # notifications_principal__notification_category=notification_category, # notifications_principal__is_enabled=True, # ).values_list("player_id", flat=True) # ) # elif principal_type == PrincipalType.EVENT_MANAGER: # # Fetch only EVENT_MANAGER principals # eligible_principals = list( # IAmPrincipal.objects.filter( # principal_type__name=PrincipalType.EVENT_MANAGER, # notifications_principal__notification_category=notification_category, # notifications_principal__is_enabled=True, # ).values_list("player_id", flat=True) # ) # return eligible_principals def get_eligible_principals_for_notification(push_notification): principal_type = push_notification.principal_type notification_category = push_notification.notification_category print("principal_type Triggered: ", principal_type) if principal_type == PrincipalType.BOTH: # If BOTH is selected, fetch users categorized under both EVENT_USER and EVENT_MANAGER print("Both Triggered") eligible_principals = [ player_id for player_id in IAmPrincipal.objects.filter( Q(principal_type__name=PrincipalType.EVENT_USER) | Q(principal_type__name=PrincipalType.EVENT_MANAGER), notifications_principal__notification_category=notification_category, notifications_principal__is_enabled=True, ).values_list("player_id", flat=True) if player_id is not None ] print("eligible_principals for both: ", eligible_principals) elif principal_type == PrincipalType.EVENT_USER: print("EVENT_USER Triggered") # Fetch only EVENT_USER principals eligible_principals = [ player_id for player_id in IAmPrincipal.objects.filter( principal_type__name=PrincipalType.EVENT_USER, notifications_principal__notification_category=notification_category, notifications_principal__is_enabled=True, ).values_list("player_id", flat=True) if player_id is not None ] print("eligible_principals for EVENT_USER: ", eligible_principals) elif principal_type == PrincipalType.EVENT_MANAGER: print("EVENT_MANAGER Triggered") # Fetch only EVENT_MANAGER principals eligible_principals = [ player_id for player_id in IAmPrincipal.objects.filter( principal_type__name=PrincipalType.EVENT_MANAGER, notifications_principal__notification_category=notification_category, notifications_principal__is_enabled=True, ).values_list("player_id", flat=True) if player_id is not None ] print("eligible_principals for EVENT_MANAGER: ", eligible_principals) return eligible_principals