我正在尝试弄清楚如何让我的应用程序使用在后台抓取网站的功能，因为它需要很长时间并且如果在前台运行会导致错误。所以我遵循了 Heroku 网站上的教程，该教程具有统计单词的功能并在后台运行。有用。所以我准备首先通过导入将我的功能放在那里。所以我导入它并创建了一个使用它的函数。我得到了这个回溯

  Traceback (most recent call last):
  File "my_raddqueue.py", line 2, in <module>
    from src.blog.my_task import conn, is_page_ok
  File "/Users/ray/Desktop/myheroku/practice/src/blog/my_task.py", line 5, in <module>
    from .my_scraps import p_panties
  File "/Users/ray/Desktop/myheroku/practice/src/blog/my_scraps.py", line 3, in <module>
    from .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.

我什至尝试在 my_task.py 中创建函数并运行它并得到相同的回溯

这是我的文件结构

下面是我认为与问题重现相关的文件和代码

我想使用的函数位于 my_scraps.py 中

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

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')), 
                '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', 
                '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

我的任务.py

    import os

    import redis
    from rq import Worker, Queue, Connection
    from .my_scraps import p_panties
    import requests


    listen = ['high', 'default', 'low']

    redis_url = os.getenv('REDISTOGO_URL', 'redis://localhost:6379')

    conn = redis.from_url(redis_url)

    if __name__ == '__main__':
        with Connection(conn):
            worker = Worker(map(Queue, listen))
            worker.work()


    def is_page_ok(url):
        response = requests.get(url)
        if response.status_code == 200:
            return "{0} is up".format(url)
        else:
            return "{0} is not OK. Status {1}".format(url, response.status_code)


    def do_this():
        a = p_panties()
        return a

我的_raddqueue.py

from rq import Queue
    from src.blog.my_task import conn, do_this

    q = Queue('important', connection=conn)

    result = q.enqueue(do_this)

    print("noted")

这一行

from .my_scraps import p_panties

即使我不使用它也会导致回溯。在我放弃尝试使用该功能后，我尝试使用并查看另一个功能是否有效，但他们没有，我无法弄清楚为什么，直到我开始删除或一个接一个地评论，当我评论或删除这一行时有效。我的问题是什么。我想要做的就是让我的应用程序在我的 heroku 应用程序中的一天中预先指定的时间抓取。我怎样才能做到这一点？我在这里的方法是错误的吗？我看到了一个叫做 APSscheduler 的东西，我应该改用它吗？任何关于改进我的代码的意见都将不胜感激。没有编码那么久。很多都是我自己的想法，所以如果它看起来不专业，那就是提前感谢你的原因

最佳答案

我不确定 Heroku。但是通常你可以通过 Celery 在 django 中实现这样的自动化任务。

这里有很棒的文档。 http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html

关于python - 我怎样才能让我的 django 应用程序在 heroku 的后台自动抓取，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/38854897/