如何为Viewflow创建新任务可用时的信号以实现Slack通知？

嘿，这个需求很实用！Viewflow默认确实只提供了像task_started这类任务启动时的信号，但要实现新任务可用状态触发的Slack通知，我们可以借助Django的内置信号机制来扩展它，下面是具体的实现步骤：

在Viewflow中，"新任务可用"通常指任务被创建后处于Task.Status.READY状态——也就是任务已经生成，但还没被分配给用户或启动执行。这是我们要监听的核心状态。

我们可以利用Django的post_save信号来监听Task模型的保存事件，判断任务是否进入了READY状态，从而触发Slack通知。

创建一个信号处理函数，比如放在你的app的signals.py文件中：

from django.db.models.signals import post_save
from django.dispatch import receiver
from viewflow.models import Task
# 假设你已经封装了发送Slack通知的工具函数
from your_app.utils import send_slack_channel_notification

@receiver(post_save, sender=Task)
def trigger_slack_notification_on_task_ready(sender, instance, created, **kwargs):
    # 处理两种情况：新创建的任务直接是READY，或者现有任务被更新为READY
    if instance.status == Task.Status.READY:
        # 可以添加过滤条件，比如只监听特定流程的任务
        # if instance.flow_class.__name__ == "YourTargetFlow":
        notification_msg = f"📢 新任务已上线：\n任务名称：{instance.flow_task.name}\n流程ID：{instance.process.id}"
        send_slack_channel_notification(channel="#task-notifications", message=notification_msg)

为了让Django加载这个信号，需要在你的app的apps.py中注册信号模块：

from django.apps import AppConfig

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app'

    def ready(self):
        # 导入信号模块，触发注册
        import your_app.signals

然后在项目的settings.py中，把你的app配置替换成这个自定义的Config：

INSTALLED_APPS = [
    # 其他app...
    'your_app.apps.YourAppConfig',
    # ...
]

如果不想每次任务保存都触发检查，我们可以只在状态从非READY变为READY时发送通知，避免重复推送：

@receiver(post_save, sender=Task)
def trigger_slack_notification_on_task_ready(sender, instance, created, **kwargs):
    if not created:
        # 获取任务更新前的状态
        prev_task = sender.objects.get(pk=instance.pk)
        if prev_task.status != Task.Status.READY and instance.status == Task.Status.READY:
            notification_msg = f"📢 新任务已上线：\n任务名称：{instance.flow_task.name}\n流程ID：{instance.process.id}"
            send_slack_channel_notification(channel="#task-notifications", message=notification_msg)
    else:
        # 新创建的任务直接是READY状态
        if instance.status == Task.Status.READY:
            notification_msg = f"📢 新任务已上线：\n任务名称：{instance.flow_task.name}\n流程ID：{instance.process.id}"
            send_slack_channel_notification(channel="#task-notifications", message=notification_msg)

启动Django服务后，创建一个流程实例，当任务进入READY状态时，去对应的Slack频道查看是否收到通知。如果没触发，可以检查Django日志，确认信号是否被正确加载和执行。

内容的提问来源于stack exchange，提问作者Tinker