超详细教程：Django 实时通知系统全栈集成方案

# settings.py

# 添加应用
INSTALLED_APPS = [
    # ... 现有应用
    'channels',  # 添加 channels
    'notifications',  # 添加 django-notifications-hq
    'anymail',  # 添加 django-anymail
    'your_app',  # 你的应用
]

# 配置 channels
ASGI_APPLICATION = 'your_project.asgi.application'  # 指定 ASGI 应用

# 配置 Redis 作为 channels 的通道层
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],  # Redis 地址和端口
        },
    },
}

# 配置 Mailgun 邮件服务
ANYMAIL = {
    "MAILGUN_API_KEY": "your-mailgun-api-key",  # Mailgun API 密钥
    "MAILGUN_SENDER_DOMAIN": "mg.yourdomain.com",  # 发件域名
}
EMAIL_BACKEND = "anymail.backends.mailgun.EmailBackend"
DEFAULT_FROM_EMAIL = "通知系统 <notifications@mg.yourdomain.com"

# 配置 django-notifications-hq
NOTIFICATIONS_BACKENDS = [
    'notifications.backends.db.DatabaseBackend',  # 存储通知到数据库
    'your_app.backends.EmailNotificationBackend',  # 自定义邮件后端（稍后创建）
]

# asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": # 稍后添加 WebSocket 路由配置
})

# your_app/backends.py
from django.core.mail import send_mail
from notifications.backends.base import BaseBackend

class EmailNotificationBackend(BaseBackend):
    def send(self, notification, *args, **kwargs):
        recipient = notification.recipient
        subject = f"【新通知】{notification.verb}"
        message = (
            f"你好，{recipient.username}！\n\n"
            f"{notification.description}\n\n"
            f"点击查看：https://yourdomain.com/notifications/"
        )
        
        # 异步发送邮件（使用 Celery 更佳，这里简化处理）
        send_mail(
            subject=subject,
            message=message,
            from_email=None,
            recipient_list=[recipient.email],
            fail_silently=False,
        )

# your_app/consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer
from asgiref.sync import sync_to_async
from django.contrib.auth.models import User

class NotificationConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        # 验证用户身份
        user = self.scope["user"]
        if user.is_anonymous:
            await self.close()
            return
            
        # 为每个用户创建独立的组
        self.group_name = f"user_{user.id}_notifications"
        
        # 将用户添加到组
        await self.channel_layer.group_add(
            self.group_name,
            self.channel_name
        )
        
        await self.accept()
        
        # 发送用户的未读通知数量
        unread_count = await self.get_unread_count(user)
        await self.send(text_data=json.dumps({
            "type": "unread_count",
            "count": unread_count
        }))
    
    async def disconnect(self, close_code):
        # 从组中移除用户
        await self.channel_layer.group_discard(
            self.group_name,
            self.channel_name
        )
    
    # 接收来自组的消息
    async def notify_message(self, event):
        message = event["message"]
        await self.send(text_data=json.dumps({
            "type": "notification",
            "data": message
        }))
    
    # 同步方法转异步
    @sync_to_async
    def get_unread_count(self, user):
        return user.notifications.unread().count()

# your_app/routing.py
from django.urls import re_path
from .consumers import NotificationConsumer

websocket_urlpatterns = [
    re_path(r'ws/notifications/$', NotificationConsumer.as_asgi()),
]

# asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import your_app.routing

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(
            your_app.routing.websocket_urlpatterns
        )
    ),
})

# your_app/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from notifications.models import Notification
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer

@receiver(post_save, sender=Notification)
def send_notification_to_websocket(sender, instance, created, **kwargs):
    if created:  # 只处理新创建的通知
        recipient = instance.recipient
        channel_layer = get_channel_layer()
        
        # 构建通知数据
        notification_data = {
            "id": instance.id,
            "verb": instance.verb,
            "description": instance.description,
            "timestamp": instance.timestamp.isoformat(),
            "unread": instance.unread,
        }
        
        # 异步发送到 WebSocket 组
        async_to_sync(channel_layer.group_send)(
            f"user_{recipient.id}_notifications",
            {
                "type": "notify_message",
                "message": notification_data
            }
        )

# your_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  # 导入信号处理

# 在 Django shell 中测试
from django.contrib.auth.models import User
from notifications import notify

sender = User.objects.get(username='admin')
recipient = User.objects.get(username='test_user')

notify.send(
    sender=sender,
    recipient=recipient,
    verb='发送了一条消息给你',
    description='请查看消息详情'
)

# your_project/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')

app = Celery('your_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# your_app/signals.py
from .tasks import send_notification_to_websocket

@receiver(post_save, sender=Notification)
def trigger_notification_tasks(sender, instance, created, **kwargs):
    if created:
        send_notification_to_websocket.delay(instance.id)

# your_app/tasks.py
from celery import shared_task
from notifications.models import Notification
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer

@shared_task
def send_notification_to_websocket(notification_id):
    try:
        notification = Notification.objects.get(id=notification_id)
        recipient = notification.recipient
        channel_layer = get_channel_layer()
        
        # 发送到 WebSocket
        async_to_sync(channel_layer.group_send)(
            f"user_{recipient.id}_notifications",
            {
                "type": "notify_message",
                "message": {
                    "id": notification.id,
                    "verb": notification.verb,
                    "description": notification.description,
                    "timestamp": notification.timestamp.isoformat(),
                }
            }
        )
    except Notification.DoesNotExist:
        pass

# your_app/backends.py
from django.core.mail import EmailMultiAlternatives
from django.template.loader import render_to_string

class EmailNotificationBackend(BaseBackend):
    def send(self, notification, *args, **kwargs):
        recipient = notification.recipient
        subject = f"【新通知】{notification.verb}"
        
        # 渲染 HTML 模板
        html_content = render_to_string('email_notification.html', {
            'user': recipient,
            'description': notification.description,
            'url': f'https://yourdomain.com/notifications/{notification.id}/'
        })
        
        # 发送邮件
        msg = EmailMultiAlternatives(
            subject=subject,
            body=notification.description,  # 纯文本备用
            to=[recipient.email]
        )
        msg.attach_alternative(html_content, "text/html")
        msg.send()