from typing import List from django.contrib.auth import get_user_model from django.db.models import Count, Q from django.shortcuts import get_object_or_404 from django.utils import timezone from ninja import Router from ninja.pagination import paginate from apps.communications.api.schemas import ( AnnouncementCreateSchema, AnnouncementListSchema, AnnouncementSchema, AnnouncementStatsSchema, MessageResponseSchema, PushDeviceCreateSchema, PushDeviceSchema, PushDeviceUpdateSchema, PushNotificationSchema, ) from apps.communications.models import Announcement, AnnouncementPriority, AnnouncementType, PushNotificationDevice from apps.communications.push_notifications import push_service from apps.communications.tasks import send_announcement_notifications from core.authentication import jwt_auth User = get_user_model() communications_router = Router() def _is_staff_user(user) -> bool: return bool(user and (user.is_staff or user.is_superuser)) @communications_router.get("/announcements/", response=List[AnnouncementListSchema]) @paginate def list_announcements(request, published_only: bool = True): queryset = Announcement.objects.select_related("author").filter(is_deleted=False) if published_only: queryset = queryset.filter(is_published=True, publish_date__lte=timezone.now()) items = [] for announcement in queryset.order_by("-created_at"): items.append( { "id": announcement.id, "title": announcement.title, "content": announcement.content, "announcement_type": announcement.announcement_type, "priority": announcement.priority, "author": announcement.author, "is_published": announcement.is_published, "publish_date": announcement.publish_date, "target_audience": announcement.target_audience, "deliver_in_app": announcement.send_email, "deliver_sms": announcement.send_push, "created_at": announcement.created_at, } ) return items @communications_router.get("/announcements/{announcement_id}/", response=AnnouncementSchema) def get_announcement(request, announcement_id: int): announcement = get_object_or_404( Announcement.objects.select_related("author").filter(is_deleted=False), id=announcement_id, ) if not announcement.is_published and not _is_staff_user(getattr(request, "auth", None)): return {"error": "Announcement not found"}, 404 return announcement @communications_router.post("/announcements/", response=AnnouncementSchema, auth=jwt_auth) def create_announcement(request, payload: AnnouncementCreateSchema): user = request.auth if not _is_staff_user(user): return {"error": "Permission denied"}, 403 announcement = Announcement.objects.create( author=user, title=payload.title, content=payload.content, announcement_type=payload.announcement_type, priority=payload.priority, target_audience=payload.target_audience, is_published=payload.is_published, publish_date=payload.publish_date, send_email=payload.deliver_in_app, send_push=payload.deliver_sms, ) if announcement.is_published and (announcement.publish_date is None or announcement.publish_date <= timezone.now()): send_announcement_notifications.delay(announcement.id) return announcement @communications_router.put("/announcements/{announcement_id}/", response=AnnouncementSchema, auth=jwt_auth) def update_announcement(request, announcement_id: int, payload: AnnouncementCreateSchema): user = request.auth if not _is_staff_user(user): return {"error": "Permission denied"}, 403 announcement = get_object_or_404(Announcement, id=announcement_id, is_deleted=False) announcement.title = payload.title announcement.content = payload.content announcement.announcement_type = payload.announcement_type announcement.priority = payload.priority announcement.target_audience = payload.target_audience announcement.is_published = payload.is_published announcement.publish_date = payload.publish_date announcement.send_email = payload.deliver_in_app announcement.send_push = payload.deliver_sms announcement.email_sent = False announcement.push_sent = False announcement.save() if announcement.is_published and (announcement.publish_date is None or announcement.publish_date <= timezone.now()): send_announcement_notifications.delay(announcement.id) return announcement @communications_router.delete("/announcements/{announcement_id}/", response=MessageResponseSchema, auth=jwt_auth) def delete_announcement(request, announcement_id: int): user = request.auth if not _is_staff_user(user): return {"error": "Permission denied"}, 403 announcement = get_object_or_404(Announcement, id=announcement_id, is_deleted=False) announcement.soft_delete() return {"message": "Announcement deleted successfully"} @communications_router.get("/announcements/stats/", response=AnnouncementStatsSchema, auth=jwt_auth) def get_announcement_stats(request): user = request.auth if not _is_staff_user(user): return {"error": "Permission denied"}, 403 stats = Announcement.objects.filter(is_deleted=False).aggregate( total_announcements=Count("id"), published_announcements=Count("id", filter=Q(is_published=True)), draft_announcements=Count("id", filter=Q(is_published=False)), urgent_announcements=Count("id", filter=Q(priority="urgent")), in_app_sent_count=Count("id", filter=Q(email_sent=True)), sms_sent_count=Count("id", filter=Q(push_sent=True)), ) return stats @communications_router.post("/newsletter/subscribe/", response=MessageResponseSchema) def subscribe_newsletter(request): return { "message": "خبرنامه ایمیلی حذف شده است. اطلاع‌رسانی‌ها از این پس درون‌سایتی و در موارد مهم از طریق پیامک انجام می‌شود." } @communications_router.post("/newsletter/unsubscribe/", response=MessageResponseSchema) def unsubscribe_newsletter(request): return {"message": "خبرنامه ایمیلی دیگر فعال نیست."} @communications_router.get("/newsletter/confirm/{token}/", response=MessageResponseSchema) def confirm_newsletter_subscription(request, token: str): return {"message": "تایید خبرنامه ایمیلی دیگر لازم نیست."} @communications_router.get("/newsletter/unsubscribe/{token}/", response=MessageResponseSchema) def unsubscribe_newsletter_token(request, token: str): return {"message": "خبرنامه ایمیلی دیگر فعال نیست."} @communications_router.post("/push-devices/", response=PushDeviceSchema, auth=jwt_auth) def register_push_device(request, payload: PushDeviceCreateSchema): user = request.auth device, created = PushNotificationDevice.objects.get_or_create( user=user, device_token=payload.device_token, defaults={"device_type": payload.device_type, "is_active": True}, ) if not created: device.is_active = True device.device_type = payload.device_type device.save() return device @communications_router.delete("/push-devices/", response=MessageResponseSchema, auth=jwt_auth) def unregister_push_device(request, device_token: str): user = request.auth try: device = PushNotificationDevice.objects.get(user=user, device_token=device_token) device.delete() return {"message": "Device unregistered successfully"} except PushNotificationDevice.DoesNotExist: return {"message": "Device not found"}, 404 @communications_router.get("/push-devices/", response=List[PushDeviceSchema], auth=jwt_auth) def list_user_push_devices(request): return PushNotificationDevice.objects.filter(user=request.auth, is_deleted=False).order_by("-created_at") @communications_router.put("/push-devices/{device_id}/", response=PushDeviceSchema, auth=jwt_auth) def update_push_device(request, device_id: int, payload: PushDeviceUpdateSchema): device = get_object_or_404(PushNotificationDevice, id=device_id, user=request.auth, is_deleted=False) device.is_active = payload.is_active device.save() return device @communications_router.post("/push-notifications/send/", response=MessageResponseSchema, auth=jwt_auth) def send_push_notification(request, payload: PushNotificationSchema): user = request.auth if not _is_staff_user(user): return {"error": "Permission denied"}, 403 users = User.objects.filter(is_active=True) if payload.target_audience == "committee": users = users.filter(is_staff=True) total_sent = push_service.send_to_multiple_users(users, payload.title, payload.body, payload.data) return {"message": f"Push notification sent to {total_sent} devices"} @communications_router.get("/announcement-types/", response=List[dict]) def get_announcement_types(request): return [{"value": choice[0], "label": choice[1]} for choice in AnnouncementType.choices] @communications_router.get("/announcement-priorities/", response=List[dict]) def get_announcement_priorities(request): return [{"value": choice[0], "label": choice[1]} for choice in AnnouncementPriority.choices]