guilan-ace-backend/apps/communications/api/views.py

from typing import List

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,
)
from apps.communications.models import Announcement, AnnouncementPriority, AnnouncementType, PushNotificationDevice
from core.api.schemas import ErrorSchema
from core.authentication import jwt_auth

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).filter(Q(publish_date__isnull=True) | Q(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,
                "created_at": announcement.created_at,
            }
        )
    return items


@communications_router.get("/announcements/stats/", response={200: AnnouncementStatsSchema, 403: ErrorSchema}, auth=jwt_auth)
def get_announcement_stats(request):
    user = request.auth
    if not _is_staff_user(user):
        return 403, {"error": "Permission denied"}
    return 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")),
    )


@communications_router.get("/announcements/{announcement_id}/", response={200: AnnouncementSchema, 404: ErrorSchema})
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 404, {"error": "Announcement not found"}
    return announcement


@communications_router.post("/announcements/", response={200: AnnouncementSchema, 403: ErrorSchema}, auth=jwt_auth)
def create_announcement(request, payload: AnnouncementCreateSchema):
    user = request.auth
    if not _is_staff_user(user):
        return 403, {"error": "Permission denied"}

    announcement = Announcement.objects.create(
        author=user,
        title=payload.title,
        content=payload.content,
        announcement_type=payload.announcement_type,
        priority=payload.priority,
        is_published=payload.is_published,
        publish_date=payload.publish_date,
    )

    return announcement


@communications_router.put("/announcements/{announcement_id}/", response={200: AnnouncementSchema, 403: ErrorSchema}, auth=jwt_auth)
def update_announcement(request, announcement_id: int, payload: AnnouncementCreateSchema):
    user = request.auth
    if not _is_staff_user(user):
        return 403, {"error": "Permission denied"}

    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.is_published = payload.is_published
    announcement.publish_date = payload.publish_date
    announcement.save()

    return announcement


@communications_router.delete("/announcements/{announcement_id}/", response={200: MessageResponseSchema, 403: ErrorSchema}, auth=jwt_auth)
def delete_announcement(request, announcement_id: int):
    user = request.auth
    if not _is_staff_user(user):
        return 403, {"error": "Permission denied"}
    announcement = get_object_or_404(Announcement, id=announcement_id, is_deleted=False)
    announcement.delete()
    return {"message": "Announcement deleted successfully"}


@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={200: MessageResponseSchema, 404: 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 404, {"message": "Device not found"}


@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.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]