Django实现WhatsApp定时消息会话：二维码会话保持问题求助

问题描述 #

我正在开发一个系统，用户除完成相关流程外，可注册客户的WhatsApp账号，用来发送自动账单、祝福等消息。用户扫码后，系统会用该WhatsApp账号定时发消息，搭建用户和客户的对话通道。目前已实现读取WhatsApp Web二维码并在模板显示，但遇到两个问题：

• 返回模板后WebDriver提前终止，导致会话无法验证
• WebDriver持续处于开启状态，无法合理释放资源

需要解决这个并发任务的资源管理问题。

当前代码 #

# views.py
from django.shortcuts import render
from django.http import HttpResponse
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import base64
import time
from django.shortcuts import render

def read_qr_code(request):
    driver = webdriver.Firefox()
    # driver.implicitly_wait(30) # mantém o webdriver ativo por 2 minutos
    driver.get('https://web.whatsapp.com/')
    wait = WebDriverWait(driver, 10)
    qr_element = wait.until(EC.presence_of_element_located((By.XPATH, '//*[@id="app"]/div/div/div[3]/div[1]/div/div/div[2]/div/canvas')))


    qr_image_binary = qr_element.screenshot_as_png
    qr_image_base64 = base64.b64encode(qr_image_binary).decode('utf-8')

    context = {
        'image_data': qr_image_base64
    }

    # send_qr(request, context)
    
    # time.sleep(20)  # aguarda por 2 minutos
    # driver.quit()  # fecha o webdriver

    return render(request, 'read_qr_code.html', context)

解决方案 #

1. 剥离WebDriver生命周期到后台任务 #

视图函数是同步请求响应模型，一旦返回模板，当前进程资源会被回收，导致WebDriver被终止。必须把WebDriver的管理逻辑放到后台异步任务中：

• 使用Celery这类异步任务框架，将WebDriver初始化、二维码获取、会话等待逻辑封装成后台任务
• 视图只负责触发任务，前端通过轮询或WebSocket获取二维码和会话验证状态
• 示例简化代码：

# tasks.py
from celery import Celery
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import base64

app = Celery('whatsapp_tasks', broker='redis://localhost:6379/0')

@app.task(bind=True)
def start_whatsapp_session(self):
    driver = webdriver.Firefox()
    driver.get('https://web.whatsapp.com/')
    try:
        # 获取二维码并更新任务状态
        qr_element = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located((By.XPATH, '//*[@id="app"]/div/div/div[3]/div[1]/div/div/div[2]/div/canvas'))
        )
        qr_base64 = base64.b64encode(qr_element.screenshot_as_png).decode('utf-8')
        self.update_state(state='QR_READY', meta={'qr_code': qr_base64})
        
        # 等待用户扫码验证，超时设为5分钟
        WebDriverWait(driver, 300).until(
            EC.presence_of_element_located((By.XPATH, '//*[@id="app"]/div/div/div[4]/div/header/div[2]/div[2]/span'))
        )
        
        # 保存会话信息，后续可复用
        session_data = {
            'cookies': driver.get_cookies(),
            'user_name': driver.find_element(By.XPATH, '//*[@id="app"]/div/div/div[4]/div/header/div[2]/div[2]/span').text
        }
        return {'status': 'SUCCESS', 'session': session_data}
    finally:
        # 验证完成后可选择关闭Driver，或保留用于后续消息发送
        # driver.quit()
        pass

# views.py
from django.shortcuts import render
from django.http import JsonResponse
from .tasks import start_whatsapp_session

def qr_page(request):
    return render(request, 'read_qr_code.html')

def start_session(request):
    if request.method == 'POST':
        task = start_whatsapp_session.delay()
        return JsonResponse({'task_id': task.id})

def check_session_status(request, task_id):
    task = start_whatsapp_session.AsyncResult(task_id)
    if task.state == 'QR_READY':
        return JsonResponse({'state': task.state, 'qr_code': task.info['qr_code']})
    elif task.state == 'SUCCESS':
        return JsonResponse({'state': task.state, 'session': task.info['session']})
    return JsonResponse({'state': task.state})

2. 会话持久化与复用 #

验证成功后，保存WebDriver的Cookie到数据库，下次需要发送消息时，直接加载Cookie恢复会话，无需重新扫码：

# 恢复会话示例
driver = webdriver.Firefox()
driver.get('https://web.whatsapp.com/')
# 从数据库读取之前保存的cookies
for cookie in saved_cookies:
    driver.add_cookie(cookie)
driver.refresh()
# 此时会话已恢复，可直接发送消息

3. 闲置资源清理 #

• 给后台任务设置超时时间，超过5分钟未验证的会话自动触发driver.quit()
• 定时扫描后台任务列表，关闭长时间闲置的WebDriver实例
• 可以用Celery的task_time_limit参数设置任务超时：

@app.task(bind=True, task_time_limit=300)
def start_whatsapp_session(self):
    # 任务逻辑

4. 替代方案：使用官方API #

如果业务场景允许，优先选择WhatsApp Business API，无需依赖Selenium模拟浏览器，稳定性和可靠性更高，也能彻底避免WebDriver的资源管理问题。

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