erpnew/notifier/tasks.py

import logging
from typing import Any

from celery import shared_task
from django.db import transaction
from django.db.models import Exists, F, OuterRef
from django.utils import timezone

from mission import models as mission_models
from mission.payload_processors import apply_mission_payload_processor
from mission import services as mission_services
from notifier.models import NotificationEventKeyEnum
from notifier.services import dispatch_notification_event

logger = logging.getLogger(__name__)


def _build_unreplied_mission_payload(*, mission: mission_models.Mission, notified_at) -> dict[str, Any]:
    participants = list(mission.get_participants().values_list("employee_id", "employee__name"))
    participant_ids = [employee_id for employee_id, _ in participants]
    participant_names = [employee_name for _, employee_name in participants]
    content_type = getattr(mission.content_type, "model", None)
    next_count = mission.unreplied_notify_sent_count + 1
    payload = {
        "mission_id": mission.id,
        "merchant_id": mission.merchant_id,
        "description": mission.description,
        "category_id": mission.category_id,
        "category_name": mission.category.name,
        "is_urgent": mission.is_urgent,
        "is_completed": mission.is_completed,
        "is_cancelled": mission.is_cancelled,
        "creator_id": mission.creator_id,
        "creator_name": getattr(mission.creator, "name", ""),
        "participant_ids": participant_ids,
        "participant_names": participant_names,
        "participant_names_display": "、".join(participant_names) if participant_names else "无",
        "content_type": content_type or "",
        "content_id": mission.content_id or "",
        "payload_processor": getattr(mission.category, "payload_processor", "") or "",
        "unreplied_notify_interval_minutes": mission.unreplied_notify_interval_minutes,
        "unreplied_notify_max_count": mission.unreplied_notify_max_count,
        "unreplied_notify_sent_count": next_count,
        "unreplied_last_notified_at": notified_at.isoformat() if notified_at else "",
    }
    return apply_mission_payload_processor(
        mission=mission,
        event_key=NotificationEventKeyEnum.MISSION_UNREPLIED,
        payload=payload,
    )


def _get_due_unreplied_mission_ids(*, limit: int) -> list[int]:
    effective_reply_exists = mission_models.MissionReply.objects.filter(
        mission_id=OuterRef("pk"),
        is_rejected=False,
    )
    queryset = (
        mission_models.Mission.objects.filter(
            notify_if_unreplied=True,
            is_completed=False,
            is_cancelled=False,
            unreplied_notify_interval_minutes__gt=0,
        )
        .filter(unreplied_notify_sent_count__lt=F("unreplied_notify_max_count"))
        .annotate(has_effective_reply=Exists(effective_reply_exists))
        .filter(has_effective_reply=False)
        .order_by("id")
    )
    return list(queryset.values_list("id", flat=True)[:limit])


def _dispatch_unreplied_notification_for_mission(*, mission_id: int) -> dict[str, Any]:
    with transaction.atomic():
        mission = (
            mission_models.Mission.objects.select_for_update(skip_locked=True)
            .select_related("category", "creator")
            .prefetch_related("participants__employee")
            .filter(id=mission_id)
            .first()
        )
        if mission is None:
            return {"mission_id": mission_id, "status": "missing"}

        if not mission.notify_if_unreplied:
            return {"mission_id": mission.id, "status": "disabled"}
        if mission.is_completed or mission.is_cancelled:
            return {"mission_id": mission.id, "status": "inactive"}
        if mission.unreplied_notify_interval_minutes is None or mission.unreplied_notify_interval_minutes <= 0:
            return {"mission_id": mission.id, "status": "invalid-interval"}
        if mission.unreplied_notify_sent_count >= mission.unreplied_notify_max_count:
            return {"mission_id": mission.id, "status": "maxed"}
        if mission.replies.filter(is_rejected=False).exists():
            return {"mission_id": mission.id, "status": "has-reply"}

        now = timezone.now()
        due_at = mission_services.get_mission_unreplied_due_at(mission=mission)
        if due_at is None or now < due_at:
            return {
                "mission_id": mission.id,
                "status": "not-due",
                "due_at": due_at.isoformat() if due_at else None,
            }

        results = dispatch_notification_event(
            event_key=NotificationEventKeyEnum.MISSION_UNREPLIED,
            merchant_id=mission.merchant_id,
            payload=_build_unreplied_mission_payload(mission=mission, notified_at=now),
        )
        if not results:
            return {"mission_id": mission.id, "status": "no-route"}

        mission.unreplied_last_notified_at = now
        mission.unreplied_notify_sent_count += 1
        mission.save(
            update_fields=[
                "unreplied_last_notified_at",
                "unreplied_notify_sent_count",
                "updated_at",
            ]
        )
        return {
            "mission_id": mission.id,
            "status": "sent",
            "results": results,
            "sent_count": mission.unreplied_notify_sent_count,
        }


@shared_task(bind=True)
def dispatch_notification_event_task(self, *, event_key: str, merchant_id: int, payload: dict | None = None) -> dict:
    results = dispatch_notification_event(
        event_key=event_key,
        merchant_id=merchant_id,
        payload=payload or {},
    )
    summary = {
        "task_id": self.request.id,
        "event_key": event_key,
        "merchant_id": merchant_id,
        "total_count": len(results),
        "sent_count": sum(1 for item in results if item.get("status") == "sent"),
        "failed_count": sum(1 for item in results if item.get("status") == "failed"),
        "results": results,
    }
    logger.info("[notifier.tasks] notification task finished: %s", summary)
    return summary


@shared_task(bind=True)
def notify_unreplied_missions_task(self, *, limit: int = 100) -> dict:
    mission_ids = _get_due_unreplied_mission_ids(limit=max(1, int(limit)))
    items = [
        _dispatch_unreplied_notification_for_mission(mission_id=mission_id)
        for mission_id in mission_ids
    ]
    summary = {
        "task_id": self.request.id,
        "candidate_count": len(mission_ids),
        "processed_count": len(items),
        "sent_count": sum(1 for item in items if item.get("status") == "sent"),
        "items": items,
    }
    logger.info("[notifier.tasks] unreplied mission notify task finished: %s", summary)
    return summary