import time from config.settings import base as settings from celery import shared_task import requests import json import logging # import firebase_admin # from firebase_admin import credentials, messaging # import firebase_admin # from firebase_admin import credentials, messaging from google.oauth2 import service_account import google.auth.transport.requests logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # تنظیمات Firebase data = { "type": "service_account", "project_id": "imamjavad-25c31", "private_key_id": "1edc90fb80a335809c4b04a713403355ff4e8bd0", "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEugIBADANBgkqhkiG9w0BAQEFAASCBKQwggSgAgEAAoIBAQCM57lia6vNNzJL\nYBwFcx49sFPXtrYKkrRhtDN13EOnF2j+y8vlwtqYR6P7HB1l10GyHx3mlN8XpYXN\n24yrTsK6WugqPdWGl/z/BgqHximH+v4NCPQTBr3lemHbTXlEkhNtmaf9zM8IzsXP\nEzD4z5u9hy3AgfZdHKO0isTvxRTuTlpTKU3PQDwiIjfb6Bk8IjfjrDQqWbHLC4am\nidwM5F+L8ecAyhVe/G7IXAflqyi4zVM/hM5FiAknA5FmyfGd9HCxaLkw3Dqtnof8\nqflJmZp0vptT3Lte7ObeUEMoRoT0bZt6DbMBI+w19OIBu0ne0OjLu/1z4CFYoR+r\nAeWGvWCHAgMBAAECgf9HnQx/FY90oO5gtiLdE/pnJxqSMtjEhufRazaDd4vOYKXD\nhLQ5EkFcsij66PnPHiZiC+BfbUpnSIAqrmsliXBSYv4OCELTJU/FovcMfHG7qtU7\nIBjsrw64ISXT+ow1+EEEAWm1eA0WwjmOBTL7CTPJA3l2QXrYu5ki8IDuP1i5UwKu\nSR3kW0+BfsQG0z2q00AjqGnFV9IuDDjcAvu2ojwanM/H+eGB+I/dtpqe87KhbBZ9\nFuKCdYNgRa3Z76mU/2jSyGQ9eyXCX0x0vKpPavkbfir7mJcvCrp+3z0h8ot0u1Mi\nj7IJd9Ot37qUj09obXyInYk8Vnj46lj8+QjdgAECgYEAxj9Fmu9oSgLBLsuYU0kU\nmUcl0HOv3UllKAYX+8Z6L/dR1KKsfSoRWQoyGE1TxXsR/uJ4uQJJZLlHMVSw3mz3\nmOHep3F5TNSM6cfJnh5/NSMoAklOzRZxW/UELcu9vaR+e1QSgBMaNmc3b483pbfs\neVD3CPPWFt2A4lI3Y77jQAECgYEAtfQLcrBYv+SEIrVML6pXrHRC19RwCzmLyC69\n07LyRG2THu26IhOK+aSzLT5FRXTOP1VD+FHfD+AOr2d1oc2HrmgxU0mVio93KSW4\nxDrmBrej1DmVjB7LSqxu7chiD/lBUdFh2Fam8dsiTQtqR02qfcQGLynvEb2yTUbj\n0lTmoIcCgYBWZ7VatgXqXBD+6FXX1v5XYB8nH4UDGb4xF5bUcclHpq/P0acEVpWB\nDWSQGwPsCpvpT6P2XvzGHcrdwV/lUfEIfUmiCV8pEWrpad6CQCCJdG03sePal3GI\n9t1/aFGmmk9WSWpWz/yYwZvzz6QdYnB638ML79rb1GccPWFO5CAAAQKBgA4+Hi9K\nEohi0N0Op/oLMXW0XA8c9/BI/uIalo1dso0crql7HljQgs5r0AK4nx+CtypJ+FoV\nvoo1lbCxPon91qMWUNYeKnCALmmwJDhoC912voI8R7KCLpOXz88ZImPxtOU8qJYQ\nolzINHUncZhHQhM6JunGNIqE+NIHvImYT709AoGALJGUb9jAg/QpSoFKlbp4xrEA\n3G/caXeB+lE19KGZxgADBbWsUsfMI7CxnROZFobCzTdhIE6N+LaAFX/6rn0P6Nf9\nN6w8//442RjkWxtmDgw7lCykXwyLSfrP3Dbzd78gGIBqngPTej9JCc7WJYnnN75M\n5TGjxvmxYqR231/L/p0=\n-----END PRIVATE KEY-----\n", "client_email": "firebase-adminsdk-fbsvc@imamjavad-25c31.iam.gserviceaccount.com", "client_id": "103207313184637638669", "auth_uri": "https://accounts.google.com/o/oauth2/auth", "token_uri": "https://oauth2.googleapis.com/token", "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs", "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/firebase-adminsdk-fbsvc%40imamjavad-25c31.iam.gserviceaccount.com", "universe_domain": "googleapis.com" } PROJECT_ID = 'imamjavad-25c31' BASE_URL = 'https://fcm.googleapis.com' SCOPES = ['https://www.googleapis.com/auth/firebase.messaging'] FCM_ENDPOINT = f'v1/projects/{PROJECT_ID}/messages:send' FCM_URL = f'{BASE_URL}/{FCM_ENDPOINT}' def _get_access_token(): """Retrieve a valid access token that can be used to authorize requests. :return: Access token. """ credentials = service_account.Credentials.from_service_account_info( data, scopes=SCOPES) request = google.auth.transport.requests.Request() credentials.refresh(request) return credentials.token # @shared_task async def send_notification(ids: list, title: str = None, body: str = None, data=None, extra_notification_kwargs: dict = None) -> list: if not ids: return [] chunked_ids = [ids[i:i + 500] for i in range(0, len(ids), 500)] responses = [] for chunk in chunked_ids: access_token = _get_access_token() headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json', } payload = { 'message': { 'token': chunk[0], 'notification': { 'title': title, 'body': body, }, 'data': {k: str(v) for k, v in (data or {}).items()}, 'android': { 'priority': 'high', 'notification': { 'title': title, 'body': body, # 'sound': 'incoming_call_sound', 'color': '#06EEBD', # 'channel_id': 'incoming_call_channel', 'visibility': 'public', }, }, } } # Send the POST request to FCM API print(f'=========(send-notif)===******') response = requests.post(FCM_URL, headers=headers, json=payload) if response.status_code == 200: logger.warning(f'Successfully sent message: {response.json()}') responses.append(response.json()) else: responses.append({'status': 'error', 'message': ""}) logger.error(f'Failed to send message notif') return responses @shared_task def send_otp_code(phone_number, code): BASE_URL_SERVICE = "https://console.melipayamak.com/api/send/simple/" phone_number = str(phone_number) code = str(code) print(code) data = {'from': '50004001410202', 'to': phone_number, 'text': code} response = requests.post(f'{BASE_URL_SERVICE}{settings.OTP_SERIVCE_KEY}', json=data) print(response.json()) def send_otp_code_whatsapp(phone_number, code): phone = phone_number if phone.startswith('0'): phone = phone[1:] phone = '98' + phone urls = [ "https://7103.api.greenapi.com/waInstance7103107557/sendMessage/dcc7cc469e274389aa3ea4d6dae9d4d126b8b07a09be41c28e", "https://7103.api.greenapi.com/waInstance7103109151/sendMessage/ed9cbea884cc49fd8032862f1bceca2074f373540dca483382", "https://7103.api.greenapi.com/waInstance7103109158/sendMessage/92d032caca1541799a4623cfcc86f449ea7f3205b30848eeab", "https://7103.api.greenapi.com/waInstance7103109163/sendMessage/d31a08b5816c432daa6e256e181274d1d334e4256d3c4555a7", ] payload = { "chatId": f"{phone}@c.us", "message": f"Habib App --aqila-- {code}" } headers = { 'Content-Type': 'application/json' } for url in urls: response = requests.request("POST", url=url, headers=headers, data=json.dumps(payload)) response.encoding = 'utf-8' response_data = response.json() invoke_status = response_data.get('invokeStatus', {}) status = invoke_status.get('status', '') print(f'>>>>>>>> {response_data}') print(f"Response: {status}") if status != "QUOTE_ALLOWED": print("OTP sent successfully.") break else: print("QUOTE_ALLOWED error, trying next URL...") time.sleep(2) @shared_task def check_live_class_reminders_task(): """ Checks for any non-cancelled live classes starting in the next 2 hours and 15 minutes and sends a reminder to all enrolled students. """ from django.utils import timezone from datetime import timedelta from apps.course.models import CourseLiveSession, Participant from apps.course.models.course import get_localized_field from apps.account.notification_service import create_and_send_notification now = timezone.now() sessions = CourseLiveSession.objects.filter( started_at__gt=now, started_at__lte=now + timedelta(hours=2, minutes=15), reminder_sent=False, is_cancelled=False ) logger.info(f"[Live Session Reminder Check] Found {sessions.count()} sessions starting within 2h 15m.") for session in sessions: participants = Participant.objects.filter(course=session.course, is_active=True).select_related('student') for p in participants: student_user = p.student course_title_en = get_localized_field('en', session.course.title) course_title_fa = get_localized_field('fa', session.course.title) create_and_send_notification( user=student_user, title_en="LIVE Class Reminder", body_en=f"The live class for '{course_title_en}' starts in 2 hours and 15 minutes.", title_fa="یادآوری کلاس زنده", body_fa=f"کلاس زنده دوره «{course_title_fa}» تا ۲ ساعت و ۱۵ دقیقه دیگر شروع می‌شود.", service='imam-javad', data={'type': 'live_class_reminder', 'session_id': session.id} ) session.reminder_sent = True session.save(update_fields=['reminder_sent']) @shared_task def check_new_course_registration_task(): """ Weekly check (Fridays) for courses with 'registering' status published in the last 7 days. Notifies all registered students about new courses. """ from django.utils import timezone from datetime import timedelta from django.contrib.auth import get_user_model from apps.course.models import Course from apps.course.models.course import get_localized_field, normalize_status from apps.account.notification_service import create_and_send_notification User = get_user_model() now = timezone.now() candidate_courses = Course.objects.filter( created_at__gte=now - timedelta(days=7), notified_new_registration=False ) new_courses = [] for course in candidate_courses: if normalize_status(course.status) == 'registering': new_courses.append(course) if not new_courses: logger.info("[New Course Registration Check] No new registering courses found in the last 7 days.") return logger.info(f"[New Course Registration Check] Found {len(new_courses)} new registering courses.") if len(new_courses) == 1: course = new_courses[0] name_en = get_localized_field('en', course.title) or "New Course" name_fa = get_localized_field('fa', course.title) or "دوره جدید" title_en = "New Course Registration" body_en = f"Registration is open for: {name_en}" title_fa = "ثبت‌نام دوره جدید" body_fa = f"ثبت‌نام برای دوره «{name_fa}» آغاز شد." elif len(new_courses) <= 3: names_en = ", ".join(get_localized_field('en', c.title) for c in new_courses[:-1]) + f", and {get_localized_field('en', new_courses[-1].title)}" names_fa = "، ".join(get_localized_field('fa', c.title) for c in new_courses[:-1]) + f" و {get_localized_field('fa', new_courses[-1].title)}" title_en = "New Courses Registration" body_en = f"Great news! Enrollment is open for: {names_en}, and others" title_fa = "ثبت‌نام دوره‌های جدید" body_fa = f"خبر خوب! ثبت‌نام برای دوره‌های «{names_fa}» و دیگر دوره‌ها آغاز شد." else: title_en = "New Courses at Imam al-Jawad School" body_en = "New directions are open at Imam al-Jawad School. Choose the right one for you!" title_fa = "دوره‌های جدید در مدرسه امام جواد" body_fa = "دوره‌های جدید در مدرسه امام جواد (ع) آغاز شده است. مسیر مناسب خود را انتخاب کنید!" students = User.objects.filter(user_type='student', is_active=True) logger.info(f"[New Course Registration Check] Notifying {students.count()} students.") for student in students: create_and_send_notification( user=student, title_en=title_en, body_en=body_en, title_fa=title_fa, body_fa=body_fa, service='imam-javad', data={'type': 'new_course_registration'} ) for course in new_courses: course.notified_new_registration = True course.save(update_fields=['notified_new_registration']) @shared_task def check_low_quiz_results_task(): """ Checks for failed quiz attempts from exactly 2, 7, or 21 days ago (score < 70) and suggests a retake to the student, if they haven't passed or retaken the quiz since. """ from django.utils import timezone from datetime import timedelta from apps.quiz.models import QuizParticipant from apps.course.models.course import get_localized_field from apps.account.notification_service import create_and_send_notification from django.core.cache import cache now = timezone.now() today = now.date() intervals = [2, 7, 21] for days in intervals: target_date = today - timedelta(days=days) # Find attempts that ended on target_date with total_score < 70 candidates = QuizParticipant.objects.filter( ended_at__date=target_date, total_score__lt=70 ).select_related('user', 'quiz') for candidate in candidates: # Check if user has retaken the quiz since this attempt has_newer_attempt = QuizParticipant.objects.filter( user=candidate.user, quiz=candidate.quiz, ended_at__gt=candidate.ended_at ).exists() # Check if user has passed the quiz at least once has_passed = QuizParticipant.objects.filter( user=candidate.user, quiz=candidate.quiz, total_score__gte=70 ).exists() if has_newer_attempt or has_passed: continue # Prevent double notification on the same day for this interval cache_key = f"notified_quiz_low_score_{candidate.id}_{days}" if not cache.get(cache_key): cache.set(cache_key, True, timeout=86400) # 24 hours quiz_title_en = get_localized_field('en', candidate.quiz.title) or f"Quiz {candidate.quiz.id}" quiz_title_fa = get_localized_field('fa', candidate.quiz.title) or f"آزمون {candidate.quiz.id}" create_and_send_notification( user=candidate.user, title_en="Quiz Retake Suggestion", body_en=f"You scored below the passing threshold on quiz '{quiz_title_en}'. We suggest taking it again to improve your score.", title_fa="پیشنهاد شرکت مجدد در آزمون", body_fa=f"امتیاز شما در آزمون «{quiz_title_fa}» کمتر از حد نصاب شده است. پیشنهاد می‌کنیم مجدداً در این آزمون شرکت کنید تا نمره خود را بهبود ببخشید.", service='imam-javad', data={'type': 'low_quiz_result', 'quiz_id': candidate.quiz.id, 'days_since_failure': days} ) @shared_task def check_inactive_students_task(): """ Finds active student users who have been registered for at least 7 days, but have no login or lesson completions in the last 7 days, and sends an inactivity alert. """ from django.utils import timezone from datetime import timedelta from django.contrib.auth import get_user_model from apps.course.models.lesson import LessonCompletion from apps.account.notification_service import create_and_send_notification from django.core.cache import cache User = get_user_model() now = timezone.now() cutoff_date = now - timedelta(days=7) # Active students registered >= 7 days ago students = User.objects.filter( user_type='student', is_active=True, date_joined__lte=cutoff_date ) logger.info(f"[Inactivity Check] Checking {students.count()} students for inactivity...") for student in students: # Check login activity in last 7 days if student.last_login and student.last_login > cutoff_date: continue # Check lesson completions in last 7 days recent_completion = LessonCompletion.objects.filter( student=student, completed_at__gt=cutoff_date ).exists() if recent_completion: continue # Check cache to prevent spamming (limit to once every 7 days) cache_key = f"notified_student_inactivity_{student.id}" if not cache.get(cache_key): cache.set(cache_key, True, timeout=86400 * 7) # 7-day cooldown create_and_send_notification( user=student, title_en="We Miss You!", body_en="You haven't taken any lessons in the last week. Come back and continue your learning journey!", title_fa="دلمان برایتان تنگ شده!", body_fa="در یک هفته گذشته هیچ درسی را مطالعه نکرده‌اید. منتظرتان هستیم تا مسیر یادگیری خود را ادامه دهید!", service='imam-javad', data={'type': 'student_inactivity'} ) @shared_task def run_scheduled_notification_task(campaign_id): """ Celery task run by Celery Beat to execute a ScheduledNotification campaign. """ from apps.account.models import ScheduledNotification from apps.account.notification_service import create_and_send_notification from django.contrib.auth import get_user_model from django.utils import timezone from datetime import timedelta campaign = ScheduledNotification.objects.filter(id=campaign_id).first() if not campaign or not campaign.is_active or not campaign.template.is_active: logger.info(f"[Scheduled Notification Task] Campaign {campaign_id} not found or disabled.") return template = campaign.template User = get_user_model() users = User.objects.filter(is_active=True) # 1. Filter target audience if campaign.target_group == ScheduledNotification.TargetGroupChoices.INACTIVE_STUDENTS: seven_days_ago = timezone.now() - timedelta(days=7) users = users.filter( user_type='student', last_login__lt=seven_days_ago ) elif campaign.target_group == ScheduledNotification.TargetGroupChoices.ALL_STUDENTS: users = users.filter(user_type='student') elif campaign.target_group == ScheduledNotification.TargetGroupChoices.ALL_PROFESSORS: users = users.filter(user_type='professor') logger.info(f"[Scheduled Notification Task] Sending campaign '{campaign.name}' to {users.count()} users...") # 2. Iterate and send for user in users: create_and_send_notification( user=user, title_en=template.title_en, body_en=template.body_en, title_ru=template.title_ru, body_ru=template.body_ru, service='imam-javad', data={'type': template.notification_type}, notification_type=template.notification_type )