06-5. Celery vs Channels 如何選擇

# ❌ 錯誤：使用 Channels（過度設計）
class NotificationConsumer(AsyncWebsocketConsumer):
    async def send_email(self, event):
        # 使用 WebSocket 發送郵件？不合適！
        send_mail(...)

# ✅ 正確：使用 Celery
from celery import shared_task

@shared_task
def send_email_task(user_id, subject, message):
    user = User.objects.get(id=user_id)
    send_mail(
        subject=subject,
        message=message,
        recipient_list=[user.email],
    )

# View 中調用
def order_view(request):
    order = Order.objects.create(...)
    send_email_task.delay(request.user.id, '訂單確認', f'訂單 {order.id} 已建立')
    return JsonResponse({'status': 'success'})

# settings.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'send-daily-report': {
        'task': 'tasks.generate_daily_report',
        'schedule': crontab(hour=2, minute=0),  # 每天凌晨 2 點
    },
    'cleanup-expired-sessions': {
        'task': 'tasks.cleanup_sessions',
        'schedule': crontab(hour=3, minute=0),  # 每天凌晨 3 點
    },
}

@shared_task(bind=True)
def export_users_to_csv(self, filters):
    """導出用戶數據到 CSV"""
    users = User.objects.filter(**filters)
    total = users.count()

    csv_buffer = io.StringIO()
    writer = csv.writer(csv_buffer)

    for idx, user in enumerate(users.iterator(), 1):
        writer.writerow([user.id, user.username, user.email])

        # 更新進度
        if idx % 100 == 0:
            self.update_state(
                state='PROGRESS',
                meta={'current': idx, 'total': total}
            )

    # 上傳到 S3
    s3_url = upload_to_s3(csv_buffer.getvalue())
    return {'url': s3_url, 'total': total}

@shared_task(bind=True, max_retries=3)
def sync_order_to_erp(self, order_id):
    """同步訂單到 ERP 系統"""
    try:
        order = Order.objects.get(id=order_id)
        response = requests.post(
            'https://erp.example.com/api/orders',
            json=order.to_dict(),
            timeout=30
        )
        response.raise_for_status()
        order.synced = True
        order.save()
    except requests.RequestException as exc:
        # 30 秒後重試
        raise self.retry(exc=exc, countdown=30)

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = f'chat_{self.room_name}'

        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )
        await self.accept()

    async def receive(self, text_data):
        data = json.loads(text_data)
        message = data['message']

        # 廣播給同房間的所有人
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message,
                'sender': self.scope['user'].username,
            }
        )

    async def chat_message(self, event):
        await self.send(text_data=json.dumps({
            'type': 'message',
            'message': event['message'],
            'sender': event['sender'],
        }))

class DocumentConsumer(AsyncWebsocketConsumer):
    async def receive(self, text_data):
        data = json.loads(text_data)

        if data['type'] == 'edit':
            # 將編輯操作廣播給其他協作者
            await self.channel_layer.group_send(
                f'doc_{self.document_id}',
                {
                    'type': 'document_edit',
                    'operation': data['operation'],
                    'position': data['position'],
                    'user': self.scope['user'].username,
                }
            )

class DashboardConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.accept()

        # 每秒推送系統狀態
        while True:
            stats = await self.get_system_stats()
            await self.send(text_data=json.dumps({
                'type': 'stats_update',
                'cpu_usage': stats['cpu'],
                'memory_usage': stats['memory'],
                'active_users': stats['users'],
            }))
            await asyncio.sleep(1)

    async def get_system_stats(self):
        return {
            'cpu': psutil.cpu_percent(),
            'memory': psutil.virtual_memory().percent,
            'users': await database_sync_to_async(
                lambda: User.objects.filter(is_active=True).count()
            )(),
        }

class GameConsumer(AsyncWebsocketConsumer):
    async def receive(self, text_data):
        data = json.loads(text_data)

        if data['type'] == 'move':
            # 玩家移動，立即廣播給其他玩家
            await self.channel_layer.group_send(
                f'game_{self.game_id}',
                {
                    'type': 'player_move',
                    'player_id': self.player_id,
                    'position': data['position'],
                    'timestamp': time.time(),
                }
            )

# views.py
def upload_video(request):
    video_file = request.FILES['video']
    video = Video.objects.create(file=video_file, status='processing')

    # 使用 Celery 處理視頻（耗時任務）
    process_video_task.delay(video.id, request.user.id)

    return JsonResponse({'video_id': video.id})

# tasks.py
@shared_task(bind=True)
def process_video_task(self, video_id, user_id):
    video = Video.objects.get(id=video_id)

    # 1. 轉碼（Celery 執行）
    for progress in range(0, 101, 10):
        time.sleep(2)  # 模擬處理
        self.update_state(state='PROGRESS', meta={'progress': progress})

        # 2. 通過 Channels 推送進度（實時通知）
        send_progress_update(user_id, video_id, progress)

    # 3. 完成後推送通知
    video.status = 'completed'
    video.save()
    send_completion_notification(user_id, video_id)

# utils.py
def send_progress_update(user_id, video_id, progress):
    from channels.layers import get_channel_layer
    from asgiref.sync import async_to_sync

    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        f'user_{user_id}',
        {
            'type': 'video_progress',
            'video_id': video_id,
            'progress': progress,
        }
    )

# consumers.py
class NotificationConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.user_id = self.scope['user'].id
        await self.channel_layer.group_add(
            f'user_{self.user_id}',
            self.channel_name
        )
        await self.accept()

    async def video_progress(self, event):
        """接收 Celery 任務推送的進度"""
        await self.send(text_data=json.dumps({
            'type': 'video_progress',
            'video_id': event['video_id'],
            'progress': event['progress'],
        }))

# views.py
def create_order(request):
    order = Order.objects.create(...)

    # 1. 同步處理：建立訂單（必須即時完成）

    # 2. 異步任務：發送郵件（Celery）
    send_order_email_task.delay(order.id)

    # 3. 實時通知：推送到前端（Channels）
    send_order_notification(request.user.id, order.id)

    return JsonResponse({'order_id': order.id})

# Webhook 回調
def payment_callback(request):
    order_id = request.POST['order_id']
    order = Order.objects.get(id=order_id)
    order.status = 'paid'
    order.save()

    # 1. 異步任務：生成發票（Celery）
    generate_invoice_task.delay(order.id)

    # 2. 實時通知：告知用戶支付成功（Channels）
    send_payment_success_notification(order.user_id, order.id)

from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync

@shared_task
def my_task(user_id):
    # 執行任務
    result = do_something()

    # 推送到 Channels
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        f'user_{user_id}',
        {
            'type': 'task_result',
            'result': result,
        }
    )