如何整合Django、Scrapy与Celery Beat并解决模块导入问题？

1. 先修复爬虫的模块导入拼写错误 #

你的spider.py里写的from scrapy_prject.items import NewsItem存在拼写错误，scrapy_prject少了字母o，应该改为：

from scrapy_project.items import NewsItem

这是导致模块找不到的直接原因之一。

2. 解决脚本运行Scrapy的路径问题 #

从根目录tasks.py直接导入Scrapy模块容易出现路径问题，更稳妥的方式是通过subprocess调用Scrapy命令行工具运行爬虫，避免模块导入报错：

from celery import shared_task
import subprocess
import os

@shared_task
def run_news_spider():
    # 切换到Scrapy项目根目录（scrapy.cfg所在路径）
    scrapy_root = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scrapy_project")
    # 执行爬虫命令，替换为你的爬虫名称
    subprocess.run(["scrapy", "crawl", "你的爬虫名字"], cwd=scrapy_root, check=True)

3. 让Scrapy能调用Django模型存入数据库 #

要让Scrapy直接使用Django的Model操作MySQL，需要在Scrapy的配置文件中初始化Django环境：
在scrapy_project/scrapy_project/settings.py开头添加：

import os
import sys

# 将Django项目根目录添加到Python路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
# 指定Django配置模块
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "news-website.settings")
# 初始化Django
import django
django.setup()

之后在pipelines.py中就可以直接导入Django模型并处理数据：

from django_app.models import NewsModel

class NewsPipeline:
    def process_item(self, item, spider):
        # 将Scrapy Item转换为Django Model实例并保存
        news_obj = NewsModel(
            title=item["title"],
            content=item["content"],
            # 对应你的Model字段
        )
        news_obj.save()
        return item

记得在Scrapy的settings.py中启用该Pipeline：

ITEM_PIPELINES = {
    "scrapy_project.pipelines.NewsPipeline": 300,
}

4. 配置Celery + Celery Beat实现定时任务 #

安装依赖 #

pip install celery redis django-celery-beat

配置Django Settings #

在news-website/settings.py中添加：

# Celery配置
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/0"
CELERY_TIMEZONE = "Asia/Shanghai" # 替换为你的时区

# 启用django-celery-beat
INSTALLED_APPS += ["django_celery_beat"]

创建Celery实例 #

在项目根目录（与manage.py同级）新建celery.py：

import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "news-website.settings")

app = Celery("news-website")
# 从Django Settings加载Celery配置，前缀为CELERY_
app.config_from_object("django.conf:settings", namespace="CELERY")
# 自动发现所有app中的tasks.py
app.autodiscover_tasks()

然后在news-website/news-website/__init__.py中添加：

from .celery import app as celery_app

__all__ = ("celery_app",)

配置定时任务（两种方式选其一） #

方式1：通过Django后台配置（推荐，灵活易修改）

先执行数据库迁移：

python manage.py migrate

启动Django后台，登录后找到Periodic Tasks菜单：

• 添加新的Periodic Task
• Task选择tasks.run_news_spider
• Interval设置为Every 24 hours
• 保存即可

方式2：代码中硬编码配置

在tasks.py中添加：

from celery.schedules import crontab
from news-website.celery import app

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # 每天凌晨0点运行爬虫
    sender.add_periodic_task(
        crontab(hour=0, minute=0),
        run_news_spider.s(),
        name="daily news crawl",
    )

启动服务 #

1. 启动Redis服务（如果用Redis作为消息中间件）
2. 启动Celery Worker：

celery -A news-website worker --loglevel=info

3. 启动Celery Beat（如果用Django后台配置定时任务，需指定调度器）：

celery -A news-website beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

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