在Django中集成snscrape脚本实现用户输入驱动的推文采集

1. 调整starter.py为可调用模块 #

先把你写的snscrape逻辑封装成可接收参数的函数，避免硬编码，方便Django调用：

# starter.py
import snscrape.modules.twitter as sntwitter
from datetime import datetime

def scrape_tweets(keyword, start_date, end_date):
    # 转换日期格式为snscrape要求的字符串格式（YYYY-MM-DD）
    start_str = start_date.strftime("%Y-%m-%d")
    end_str = end_date.strftime("%Y-%m-%d")
    query = f"{keyword} since:{start_str} until:{end_str}"
    
    tweets = []
    try:
        for i, tweet in enumerate(sntwitter.TwitterSearchScraper(query).get_items()):
            # 按需提取推文字段，比如内容、发布时间、作者等
            tweets.append({
                'content': tweet.content,
                'date': tweet.date.strftime("%Y-%m-%d %H:%M:%S"),
                'username': tweet.user.username
            })
            # 可设置采集数量上限，避免无限采集
            if i > 100:
                break
        return {'status': 'success', 'data': tweets}
    except Exception as e:
        return {'status': 'error', 'message': str(e)}

2. 编写Django表单与视图 #

表单类（在app的forms.py中） #

创建表单接收用户输入的关键词和日期范围：

# forms.py
from django import forms
from datetime import date

class TweetScrapeForm(forms.Form):
    keyword = forms.CharField(label='搜索关键词', max_length=100)
    start_date = forms.DateField(label='开始日期', widget=forms.DateInput(attrs={'type': 'date'}), initial=date.today())
    end_date = forms.DateField(label='结束日期', widget=forms.DateInput(attrs={'type': 'date'}), initial=date.today())

视图函数（在app的views.py中） #

处理表单提交，调用starter.py的采集函数，注意：直接同步采集会导致请求超时，这里分两种方案：

方案A：简单同步调用（仅用于测试，不适合生产）

# views.py
from django.shortcuts import render
from .forms import TweetScrapeForm
from .starter import scrape_tweets

def tweet_scraper(request):
    results = None
    if request.method == 'POST':
        form = TweetScrapeForm(request.POST)
        if form.is_valid():
            keyword = form.cleaned_data['keyword']
            start_date = form.cleaned_data['start_date']
            end_date = form.cleaned_data['end_date']
            results = scrape_tweets(keyword, start_date, end_date)
    else:
        form = TweetScrapeForm()
    return render(request, 'tweet_scraper.html', {'form': form, 'results': results})

方案B：异步任务处理（生产环境推荐）

用django-background-tasks库实现异步采集（无需复杂的Celery配置）：

1. 安装依赖：pip install django-background-tasks
2. 在settings.py添加'background_task'到INSTALLED_APPS
3. 创建异步任务（在app的tasks.py中）：

# tasks.py
from background_task import background
from .starter import scrape_tweets
# 生产环境建议将结果存入数据库，此处用全局变量仅作测试示例
scrape_results = {}

@background(schedule=1)
def run_scrape_task(task_id, keyword, start_date, end_date):
    result = scrape_tweets(keyword, start_date, end_date)
    scrape_results[task_id] = result

4. 修改视图函数：

# views.py
from django.shortcuts import render
from .forms import TweetScrapeForm
from .tasks import run_scrape_task, scrape_results
import uuid

def tweet_scraper(request):
    task_id = None
    results = None
    if request.method == 'POST':
        form = TweetScrapeForm(request.POST)
        if form.is_valid():
            keyword = form.cleaned_data['keyword']
            start_date = form.cleaned_data['start_date']
            end_date = form.cleaned_data['end_date']
            task_id = str(uuid.uuid4())
            # 提交异步任务
            run_scrape_task(task_id, keyword, start_date, end_date)
    else:
        form = TweetScrapeForm()
    
    # 检查任务是否完成
    if request.GET.get('task_id'):
        task_id = request.GET.get('task_id')
        results = scrape_results.get(task_id)
    
    return render(request, 'tweet_scraper.html', {'form': form, 'task_id': task_id, 'results': results})

3. 编写前端模板（templates/tweet_scraper.html） #

<!DOCTYPE html>
<html>
<head>
    <title>推文采集工具</title>
</head>
<body>
    <h1>搜索推文</h1>
    <form method="post">
        {% csrf_token %}
        {{ form.as_p }}
        <button type="submit">开始采集</button>
    </form>

    {% if task_id %}
        <p>采集任务已启动，<a href="?task_id={{ task_id }}">点击刷新查看结果</a></p>
    {% endif %}

    {% if results %}
        {% if results.status == 'success' %}
            <h2>采集结果（共{{ results.data|length }}条）</h2>
            <ul>
                {% for tweet in results.data %}
                    <li>
                        <strong>{{ tweet.username }}</strong> {{ tweet.date }}<br>
                        {{ tweet.content }}
                    </li>
                {% endfor %}
            </ul>
        {% else %}
            <p style="color:red;">采集失败：{{ results.message }}</p>
        {% endif %}
    {% endif %}
</body>
</html>

4. 配置URL路由 #

在app的urls.py中添加路由：

# urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('scrape/', views.tweet_scraper, name='tweet_scraper'),
]

注意事项 #

• 生产环境必须用异步任务，否则长耗时的采集会导致请求超时
• 建议将采集结果存入Django模型（比如创建Tweet模型），替代示例中的全局变量
• 注意snscrape的使用限制，避免频繁请求被Twitter限制
• 添加日志记录，方便排查采集过程中的问题

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