from django.conf import settings from django.core.files.uploadedfile import UploadedFile from django.core.mail import EmailMessage from django.utils.html import strip_tags from django.template.loader import render_to_string from django.shortcuts import get_object_or_404 from smtplib import SMTPException from module_iam.models import IAmPrincipal, IAmPrincipalOtp, IAmPrincipalType from .utils import RandomGenerator # from twilio.rest import Client from django.db.models import Q import phonenumbers from decimal import Decimal from django.db.models import F from django.db import transaction from datetime import timedelta, time, datetime from django.utils import timezone import requests from onesignal_sdk.client import Client as OneSignalClient import logging import onesignal from onesignal.models import Notification from onesignal.api import default_api logger = logging.getLogger(__name__) class EmailService: email = None body = None subject = None to = None from_email = None content_subtype = "html" def __init__(self, subject=None, to=None, from_email=None): self.subject = subject self.to = (to,) self.from_email = from_email def set_to(self, to): self.to = to def set_subject(self, subject): self.subject = subject def set_from_email(self, from_email): self.from_email = from_email def set_text_body(self, body): self.body = strip_tags(body) def set_html_body(self, html_body): self.body = html_body def load_template(self, path=None, context={}): if path is None: raise Exception("Email temaplate path is not provided.") self.content_subtype = "html" html_body = render_to_string(path, context=context) self.body = html_body def attach(self, file_path): self.email.attach_file(file_path) def send(self): try: self.email = EmailMessage( subject=self.subject, body=self.body, to=self.to, from_email=self.from_email, ) self.email.content_subtype = self.content_subtype self.email.send() except SMTPException as e: logger.error(str(e)) class SMSError(Exception): def __init__(self, message, payload=None): self.message = message self.payload = payload def __str__(self): return str(self.message) class SMSService: # def send(self, to: list, text: str): # """ # Sends text sms to the given user(s). # Parameters: # to (list): list of phone numbers # text (str): a text message. # Return: # True or False # """ # # if settings.TESTING_ENV: # # logger.info(f"TESTING ENV SMS LOG : {text}") # # return True # account_sid = settings.TWILIO_ACCOUNT_SID # auth_token = settings.TWILIO_AUTH_TOKEN # my_twilio_number = settings.MY_TWILIO_NUMBER # client = Client(account_sid, auth_token) # try: # for number in to: # logger.info("SENDING SMS TO " + str(number)) # message = client.messages.create( # from_=my_twilio_number, body=text, to=number # ) # except Exception as e: # logger.error(str(e)) # raise SMSError(message=str(e)) def create_otp(self, principal: IAmPrincipal, otp_purpose: str): old_otp_change = IAmPrincipalOtp.objects.filter(principal=principal).update(is_used=True) otp = IAmPrincipalOtp.objects.create( principal=principal, otp_purpose=otp_purpose ) otp.save() return otp.otp_code def send_otp(self, principal: IAmPrincipal, otp_purpose: str): """ Sends otp to the given user. Parameters: user (User): User object otp_purpose (str) : a text that describe otp purpose Return: True or False """ if not isinstance(principal, IAmPrincipal): raise Exception( f"parameter 'principal' required type of User object, Given {type(principal)} type object" ) otp_code = self.create_otp(principal=principal, otp_purpose=otp_purpose) # below working will change as it is temporary purpose body = f"Your Nifty11 OTP is {otp_code}." print(body) phone_numbers = [] try: parsed_number = phonenumbers.parse(str(principal.phone_no), None) if phonenumbers.is_valid_number(parsed_number): formatted_number = phonenumbers.format_number( parsed_number, phonenumbers.PhoneNumberFormat.E164 ) phone_numbers.append(formatted_number) else: raise ValueError("Invalid phone number") except Exception as e: logger.warning(f"{e}") raise ValueError("Invalid phone number") if not phone_numbers: raise ValueError("Invalid phone number") print(f"phone number {type(phone_numbers)} {phone_numbers}") # self.send(phone_numbers, body) return otp_code # by using office onesignal package onesignal-python-api class OneSignalService: def __init__(self): # Get the OneSignal app key and user key from the environment variables self.configuration = onesignal.Configuration( app_key=settings.ONESIGNAL_APP_ID, api_key=settings.ONESIGNAL_REST_API_KEY ) # Create an instance of the OneSignal API self.api_client = onesignal.ApiClient(self.configuration) self.api_instance = default_api.DefaultApi(self.api_client) def send_notification(self, headings, contents, include_player_ids=None): # Create a notification object using a dictionary notification = Notification( app_id=self.configuration.app_key, include_player_ids=include_player_ids, headings={"en": headings}, contents={"en": contents} ) try: # Send the notification response = self.api_instance.create_notification( notification=notification, async_req=True ) except Exception as e: raise Exception("Generic OneSignal error: {}".format(e)) print("complete service is succeesss") return response # by using community packgae onesignal-sdk class OneSignalNotificationService: """ Class for sending notifications using the OneSignal API. Provides a convenient way to create and send notifications to OneSignal users, with features like targeting specific devices or segments, customizing notification content, and handling errors gracefully. **Parameters:** - **app_id** (str): Your OneSignal App ID. - **rest_api_key** (str): Your OneSignal REST API Key. - **user_auth_key** (str): Your OneSignal User Auth Key. **Keyword Arguments:** This method accepts additional keyword arguments (`**kwargs`) to customize the notification further, including: - `url` (str): URL to open when the notification is clicked. - `data` (dict): Custom data to be sent with the notification. - `buttons` (list): List of action buttons to display within the notification. - `send_after` (str): Timestamp for scheduling the notification. - `delayed_option` (dict): Option for delayed delivery (Android-specific). - `android_channel_id` (str): Channel ID for Android notifications. - `ios_sound` (str): Sound to play for iOS notifications. - `ios_badgeType` (str): Badge type for iOS notifications. - `ios_badgeCount` (int): Badge count for iOS notifications. - `ios_thread_id` (str): Thread ID to group notifications in iOS. - `android_background_layout` (str): Layout for background notifications on Android. - `android_group` (str): Group notification on Android. - `android_group_message` (str): Summary for grouped notifications on Android. - `android_group_summary` (str): Summary for grouped notifications on Android. - `android_led_color` (str): LED color for Android notifications. - `android_accent_color` (str): Accent color for Android notifications. - `android_visibility` (str): Visibility settings for Android notifications. **Example usage:** notification = OneSignalNotificationService() response = notification.send_notification( headings="Welcome", message="Thanks for signing up!", player_tokens=["PLAYER_TOKEN1", "PLAYER_TOKEN2"], url="https://yourwebsite.com/welcome", data={"user_id": 123}, ) """ def __init__(self): self.config = OneSignalClient( app_id=settings.ONESIGNAL_APP_ID, rest_api_key=settings.ONESIGNAL_REST_API_KEY, user_auth_key=settings.ONESIGNAL_USER_AUTH_KEY ) # Set up logging self.logger = logging.getLogger(__name__) def send_notification(self, headings, message, player_tokens=None, **kwargs): notification_obj = { "headings": {"en": headings}, "contents": {"en": message}, **kwargs } if player_tokens: notification_obj["include_player_ids"] = player_tokens try: response = self.config.send_notification(notification_obj) self.logger.info(f"Notification send successfully : {response}") return response except Exception as e: self.logger.error(f"OneSignal error {e}") raise Exception("Generic OneSignal error: {}".format(e))