forked from erp-dev/erp
feat: mission replied calling task
This commit is contained in:
@@ -1,12 +1,124 @@
|
||||
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 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]:
|
||||
participant_names = list(
|
||||
mission.get_participants().values_list("employee__name", flat=True)
|
||||
)
|
||||
content_type = getattr(mission.content_type, "model", None)
|
||||
next_count = mission.unreplied_notify_sent_count + 1
|
||||
return {
|
||||
"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_names": participant_names,
|
||||
"participant_names_display": "、".join(participant_names) if participant_names else "无",
|
||||
"content_type": content_type or "",
|
||||
"content_id": mission.content_id 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 "",
|
||||
}
|
||||
|
||||
|
||||
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(
|
||||
@@ -25,3 +137,21 @@ def dispatch_notification_event_task(self, *, event_key: str, merchant_id: int,
|
||||
}
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user