使用django、芹菜和redis异步抓取和存储我的结果并存储我的结果的正确方法是什么？

Question

当我尝试使用我在django应用程序中创建的函数时，我一直在试图理解我的问题是什么。该功能转到一个网站，收集数据并将其存储在我的数据库中。起初，我尝试使用rq和redis一段时间，但我一直收到错误消息。所以有人认为我应该试着用芹菜，我就试了。但我现在明白了rq和芹菜才是问题所在。因为我得到了和以前一样的错误信息。我厌倦了导入它，但仍然得到了错误消息，然后我想，如果我在tasks.py文件中有实际的函数，它会产生不同的效果，但它没有。

import requests
from bs4 import BeautifulSoup
from src.blog.models import Post
import random
import re
from django.contrib.auth.models import User
import os

@app.tasks
def p_panties():
    def swappo():
        user_one = ' "Mozilla/5.0 (Windows NT 6.0; WOW64; rv:24.0) Gecko/20100101 Firefox/24.0" '
        user_two = ' "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_5)" '
        user_thr = ' "Mozilla/5.0 (Windows NT 6.3; Trident/7.0; rv:11.0) like Gecko" '
        user_for = ' "Mozilla/5.0 (Macintosh; Intel Mac OS X x.y; rv:10.0) Gecko/20100101 Firefox/10.0" '

        agent_list = [user_one, user_two, user_thr, user_for]
        a = random.choice(agent_list)
        return a

    headers = {
        "user-agent": swappo(),
        "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "accept-charset": "ISO-8859-1,utf-8;q=0.7,*;q=0.3",
        "accept-encoding": "gzip,deflate,sdch",
        "accept-language": "en-US,en;q=0.8",
    }

    pan_url = 'http://www.example.org'
    shtml = requests.get(pan_url, headers=headers)
    soup = BeautifulSoup(shtml.text, 'html5lib')
    video_row = soup.find_all('div', {'class': 'post-start'})
    name = 'pan videos'

    if os.getenv('_system_name') == 'OSX':
        author = User.objects.get(id=2)
    else:
        author = User.objects.get(id=3)

    def youtube_link(url):
        youtube_page = requests.get(url, headers=headers)
        soupdata = BeautifulSoup(youtube_page.text, 'html5lib')
        video_row = soupdata.find_all('p')[0]
        entries = [{'text': div,
                    } for div in video_row]
        tubby = str(entries[0]['text'])
        urls = re.findall('http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', tubby)
        cleaned_url = urls[0].replace('?&autoplay=1', '')
        return cleaned_url

    def yt_id(code):
        the_id = code
        youtube_id = the_id.replace('https://www.youtube.com/embed/', '')
        return youtube_id

    def strip_hd(hd, move):
        str = hd
        new_hd = str.replace(move, '')
        return new_hd

    entries = [{'href': div.a.get('href'),
                'text': strip_hd(strip_hd(div.h2.text, '– Official video HD'), '– Oficial video HD').lstrip(),
                'embed': youtube_link(div.a.get('href')), #embed
                'comments': strip_hd(strip_hd(div.h2.text, '– Official video HD'), '– Oficial video HD').lstrip(),
                'src': 'https://i.ytimg.com/vi/' + yt_id(youtube_link(div.a.get('href'))) + '/maxresdefault.jpg', #image
                'name': name,
                'url': div.a.get('href'),
                'author': author,
                'video': True

                } for div in video_row][:13]

    for entry in entries:
        post = Post()
        post.title = entry['text']
        title = post.title
        if not Post.objects.filter(title=title):
            post.title = entry['text']
            post.name = entry['name']
            post.url = entry['url']
            post.body = entry['comments']
            post.image_url = entry['src']
            post.video_path = entry['embed']
            post.author = entry['author']
            post.video = entry['video']
            post.status = 'draft'
            post.save()
            post.tags.add("video", "Musica")
    return entries

在python shell中，如果我运行

from tasks import *

我得到了

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/ray/Desktop/myheroku/practice/tasks.py", line 5, in <module>
    from src.blog.models import Post
  File "/Users/ray/Desktop/myheroku/practice/src/blog/models.py", line 3, in <module>
    from taggit.managers import TaggableManager
  File "/Users/ray/Desktop/myheroku/practice/lib/python3.5/site-packages/taggit/managers.py", line 7, in <module>
    from django.contrib.contenttypes.models import ContentType
  File "/Users/ray/Desktop/myheroku/practice/lib/python3.5/site-packages/django/contrib/contenttypes/models.py", line 159, in <module>
    class ContentType(models.Model):
  File "/Users/ray/Desktop/myheroku/practice/lib/python3.5/site-packages/django/contrib/contenttypes/models.py", line 160, in ContentType
    app_label = models.CharField(max_length=100)
  File "/Users/ray/Desktop/myheroku/practice/lib/python3.5/site-packages/django/db/models/fields/__init__.py", line 1072, in __init__
    super(CharField, self).__init__(*args, **kwargs)
  File "/Users/ray/Desktop/myheroku/practice/lib/python3.5/site-packages/django/db/models/fields/__init__.py", line 166, in __init__
    self.db_tablespace = db_tablespace or settings.DEFAULT_INDEX_TABLESPACE
  File "/Users/ray/Desktop/myheroku/practice/lib/python3.5/site-packages/django/conf/__init__.py", line 55, in __getattr__
    self._setup(name)
  File "/Users/ray/Desktop/myheroku/practice/lib/python3.5/site-packages/django/conf/__init__.py", line 41, in _setup
    % (desc, ENVIRONMENT_VARIABLE))
django.core.exceptions.ImproperlyConfigured: Requested setting DEFAULT_INDEX_TABLESPACE, but settings are not configured. You must either define the environment variable DJANGO_SETTINGS_MODULE or call settings.configure() before accessing settings.

这和我用rq和redis得到的追踪完全一样。我发现如果我像这样修改进口

import requests
from bs4 import BeautifulSoup
# from src.blog.models import Post
import random
import re
# from django.contrib.auth.models import User
import os

然后像这样修改我的函数

@app.task
def p_panties():
    def swappo():
        user_one = ' "Mozilla/5.0 (Windows NT 6.0; WOW64; rv:24.0) Gecko/20100101 Firefox/24.0" '
        user_two = ' "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_5)" '
        user_thr = ' "Mozilla/5.0 (Windows NT 6.3; Trident/7.0; rv:11.0) like Gecko" '
        user_for = ' "Mozilla/5.0 (Macintosh; Intel Mac OS X x.y; rv:10.0) Gecko/20100101 Firefox/10.0" '

        agent_list = [user_one, user_two, user_thr, user_for]
        a = random.choice(agent_list)
        return a

    headers = {
        "user-agent": swappo(),
        "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "accept-charset": "ISO-8859-1,utf-8;q=0.7,*;q=0.3",
        "accept-encoding": "gzip,deflate,sdch",
        "accept-language": "en-US,en;q=0.8",
    }

    pan_url = 'http://www.example.org'
    shtml = requests.get(pan_url, headers=headers)
    soup = BeautifulSoup(shtml.text, 'html5lib')
    video_row = soup.find_all('div', {'class': 'post-start'})
    name = 'pan videos'

    # if os.getenv('_system_name') == 'OSX':
    #     author = User.objects.get(id=2)
    # else:
    #     author = User.objects.get(id=3)

    def youtube_link(url):
        youtube_page = requests.get(url, headers=headers)
        soupdata = BeautifulSoup(youtube_page.text, 'html5lib')
        video_row = soupdata.find_all('p')[0]
        entries = [{'text': div,
                    } for div in video_row]
        tubby = str(entries[0]['text'])
        urls = re.findall('http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', tubby)
        cleaned_url = urls[0].replace('?&autoplay=1', '')
        return cleaned_url

    def yt_id(code):
        the_id = code
        youtube_id = the_id.replace('https://www.youtube.com/embed/', '')
        return youtube_id

    def strip_hd(hd, move):
        str = hd
        new_hd = str.replace(move, '')
        return new_hd

    entries = [{'href': div.a.get('href'),
                'text': strip_hd(strip_hd(div.h2.text, '– Official video HD'), '– Oficial video HD').lstrip(),
                'embed': youtube_link(div.a.get('href')), #embed
                'comments': strip_hd(strip_hd(div.h2.text, '– Official video HD'), '– Oficial video HD').lstrip(),
                'src': 'https://i.ytimg.com/vi/' + yt_id(youtube_link(div.a.get('href'))) + '/maxresdefault.jpg', #image
                'name': name,
                'url': div.a.get('href'),
                # 'author': author,
                'video': True

                } for div in video_row][:13]
    #
    # for entry in entries:
    #     post = Post()
    #     post.title = entry['text']
    #     title = post.title
    #     if not Post.objects.filter(title=title):
    #         post.title = entry['text']
    #         post.name = entry['name']
    #         post.url = entry['url']
    #         post.body = entry['comments']
    #         post.image_url = entry['src']
    #         post.video_path = entry['embed']
    #         post.author = entry['author']
    #         post.video = entry['video']
    #         post.status = 'draft'
    #         post.save()
    #         post.tags.add("video", "Musica")
    return entries

它能工作，因为这是我的输出

[2016-08-13 08:31:17,222: INFO/MainProcess] Received task: tasks.p_panties[e196c6bf-2b87-4bb2-ae11-452e3c41434f]
[2016-08-13 08:31:17,238: INFO/Worker-4] Starting new HTTP connection (1): www.example.org
[2016-08-13 08:31:17,582: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:18,314: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:18,870: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:19,476: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:20,089: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:20,711: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:21,218: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:21,727: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:22,372: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:22,785: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:23,375: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:23,983: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:24,396: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:25,003: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:25,621: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:26,029: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:26,446: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:27,261: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:27,671: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:28,082: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:28,694: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:29,311: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:29,922: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:30,535: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:31,154: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:31,765: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:32,387: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:32,992: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:33,611: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:34,030: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:34,635: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:35,041: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:35,659: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:36,278: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:36,886: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:37,496: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:37,913: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:38,564: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:39,143: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:39,754: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:40,409: INFO/Worker-4] Starting new HTTP connection (1): example.org
[2016-08-13 08:31:40,992: INFO/MainProcess] Task tasks.p_panties[e196c6bf-2b87-4bb2-ae11-452e3c41434f] succeeded in 23.767645187006565s: [{'src': 'https://i.ytimg.com/vi/3bU-AtShW7Y/maxresdefault.jpg', 'name': 'pan videos', 'url':...

似乎需要某种类型的授权才能与我的Post模型交互。我只是不知道怎么做。我一直在网上寻找如何抓取数据并将数据保存到数据库中的示例。奇怪的是，我从来没有遇到过。我能读到的任何建议都会有很大的帮助。

编辑

我的文件结构

environ\
  |-src\
     |-blog\
        |-migrations\
        |-static\
        |-templates\
        |-templatetags\
        |-__init__.py
        |-admin.py
        |-forms.py
        |-models
        |-tasks
        |-urls
        |-views

Answer 1

你需要设置Django

您似乎试图在Python 中运行您的任务，这更有可能是因为您的代码在注释掉Django模型部分时工作正常。

因此，问题是，当运行纯python时，Django需要安装才能运行良好。当您在manage.py shell中运行它时，manage.py会照顾或设置它，但是通过python脚本执行它需要手动设置。这是缺少DJANGO_SETTINGS_MODULE错误的原因。

您似乎还使用了您定义的模型，为了能够将它们导入到您的python脚本中，需要将路径添加到项目的根文件夹到当前的python 中。

最后，您需要告诉django您的设置文件在哪里(在设置django之前)，在manage.py文件中，您应该有如下内容：

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myapp.settings")

将其设为常量，将其命名为DEFAULT_SETTINGS_MODULE，因此您现在拥有：

os.environ.setdefault("DJANGO_SETTINGS_MODULE", DEFAULT_SETTINGS_MODULE)

现在，您需要将常量导入到脚本中，并告诉django (通过设置env )它应该在哪里查找设置文件。

因此，总的来说：

import sys, os
sys.path.insert(0, "/path/to/parent/of/src") # /home/projects/my-crawler

from manage import DEFAULT_SETTINGS_MODULE
os.environ.setdefault("DJANGO_SETTINGS_MODULE", DEFAULT_SETTINGS_MODULE)

import django
django.setup() 
... The rest of your script ...

这样你就可以安顿好了。但是，如果您想要运行一个芹菜任务，您应该使用.delay()或.apply_async()，以确保代码在后台运行。

我自己的建议是使用python manage.py 运行python，这种情况下django为您处理所有事情。您只需要导入任务并运行它。

此外，关于存储您的刮取任务的结果，您可以在数据库中，或者在redis中，或者在任何地方(一个文件，另一个web服务器.等等，您也可以调用另一个芹菜任务来处理结果并将条目传递给它)。

只需将其添加到任务代码的末尾即可。

Redis

from redis import StrictRedis

redis = StrictRedis(host='localhost', port=6379, db=0)

redis.set("scraping:tasks:results:TASK-ID-HERE", json.dumps(entries))

这是保存结果的最简单方法，但也可以使用Redis 清单/地图。

仅供参考，这是你使用列表的方式

with redis.pipeline() as pipe:
    for item in entries:
        pipe.rpush("scraping:tasks:results", json.dumps(item))
    pipe.execute()

-编辑

正如我已经提到的，您可以定义另一个芹菜任务来处理当前刮擦的结果。所以基本上你有以下几点：

@celery_app.task
def handle_scraping_results(entries):
    you do whatever you want with the entries array now

并在p_panties任务结束时调用它，如下所示：

handle_scraping_results.delay(entries)

RabbitMQ在这里所做的就是将消息从p_panties任务传递到任务。您需要注意的是，这些不是简单的函数，共享相同的内存地址空间，它们可以在不同的进程中，在不同的服务器上！实际上这就是芹菜的用途。您不能调用处于不同过程中的函数。RabbitMQ从进程A(具有任务p_panties)获取消息，并将其传递给进程B(具有任务handle_result) (消息传递是RPC的完美方法)。

你不能在狂犬病里拯救任何东西，这不像红狗。我鼓励您在芹菜上阅读更多，因为您似乎选择了错误的基础。使用芹菜不会解决你的问题，它实际上增加了它(因为它可能很难理解在一开始)。如果您不需要异步处理，只需完全摆脱芹菜。让您的代码是一个单一的函数，您可以像我前面所描述的那样，轻松地从python或manage.py shell调用它。

--编辑II

您希望每隔几个小时在DB中保持一次。因此，无论任务完成与否，结果都会丢失，所以您必须坚持。

你有两个选择

1. 每当任务完成时(这不是每几个小时一次)，就在DB中持久化。
2. 每当任务完成时，要在Redis中持久化，然后每隔几个小时就有一个周期任务，将它们保存在django数据库中。

第一种方法很简单，您只需取消注释您在自己代码中注释过的代码。第二条路需要更多的工作。

考虑到您的结果是在redis中持久化的，正如我告诉您的那样，您可以有如下所示的定期任务来为您处理持久化DB。

redis_keys = redis.get("scraping:tasks:results:*")

for key in redis_keys:
    value_of_redis_key = redis.get(key)
    entries = json.loads(entries)
    for entry in entries:
        post = Post()
        post.title = entry['text']
        title = post.title
        if not Post.objects.filter(title=title):
            post.title = entry['text']
            post.name = entry['name']
            post.url = entry['url']
            post.body = entry['comments']
            post.image_url = entry['src']
            post.video_path = entry['embed']
            post.author = entry['author']
            post.video = entry['video']
            post.status = 'draft'
            post.save()
            post.tags.add("video", "Musica")