用 Python 構建內容聚合器

目錄

演示：您將構建的內容

項目概況

先決條件

第 1 步：設置您的項目

第 2 步：構建您的播客模型

第 3 步：創建您的主頁視圖

第 4 步：解析 Podcast RSS 提要

第 5 步：創建 Django 自定義命令

第 6 步：向 Python 內容聚合器添加其他提要

第 7 步：使用 django-apscheduler 調度任務

結論

下一步

在這個基于項目的教程中，您將使用 Python 和流行的框架Django從頭開始構建一個內容聚合器。

由于每天都有大量內容在線發布，因此訪問多個站點和來源以獲取有關您最喜歡的主題的信息可能會非常耗時。這就是內容聚合器如此受歡迎和強大的原因，因為您可以使用它們在一個地方查看所有最新新聞和內容。

無論您是在尋找投資組合項目，還是在尋找將未來項目擴展到簡單CRUD功能之外的方法，本教程都會為您提供幫助。

在本教程中，您將學習：

如何使用RSS 提要

如何創建Django自定義管理命令

如何按計劃自動運行自定義命令

如何使用單元測試來測試 Django 應用程序的功能

單擊下面的鏈接下載此項目的代碼，并按照以下步驟構建自己的內容聚合器：

演示：您將構建的內容

您將在名為pyCasts 的Python 中構建您自己的播客內容聚合器！從頭到尾遵循本教程。

該應用程序將是一個網頁，顯示來自The Real Python Podcast和Talk Python to Me Podcast的最新 Python Podcast 劇集。完成本教程后，您可以通過向應用程序添加更多播客提要來練習所學。

這是一個快速演示視頻，展示了它的實際效果：

項目概況

為了能夠向最終用戶顯示內容，您需要遵循以下幾個步驟：

設置項目

構建播客模型

創建主頁視圖

解析播客 RSS 提要

創建 Django 自定義命令

添加其他提要

使用 django-apscheduler 調度任務

在本教程的過程中，您將逐步了解其中的每一個。現在，您將了解將在上述步驟中使用哪些技術和框架。

為了將播客 RSS 提要提取到您的應用程序中并對其進行解析，您將學習如何使用feedparser庫。您將使用該庫僅從提要中提取最新的劇集數據，您將把這些數據編組到Episode模型中并使用 Django ORM 保存到數據庫中。

您可以將此代碼添加到腳本中并定期手動運行它，但這會破壞使用聚合器來節省時間的意義。相反，您將學習如何使用稱為自定義管理命令的內置 Django 工具。要解析和保存數據，您將在 Django 內部運行您的代碼。

在django-apscheduler庫的幫助下，您將為您的函數調用設置時間表，也稱為作業。然后，您可以使用 Django 管理面板查看運行的作業和時間。這將確保在不需要管理員干預的情況下自動獲取和解析您的提要。

然后，您將使用Django 模板引擎向用戶顯示查詢的上下文 — 換句話說，最新的劇集。

先決條件

為了充分利用本教程，您應該熟悉以下概念和技術：

Python基礎

虛擬環境設置和使用

基本的HTML和CSS

Django 基礎知識，例如其文件夾結構、URL 路由、遷移以及如何創建項目和應用程序

您可能還會發現有一些Bootstrap 4 的經驗會很有幫助。

如果您在開始本教程之前沒有掌握所有必備知識，那也沒關系！事實上，您可以通過繼續前進和剛剛開始來了解更多信息。如果遇到困難，您可以隨時停下來查看上面鏈接的資源。

第 1 步：設置您的項目

在這一步結束時，您將設置環境、安裝依賴項并完成 Django 的啟動和運行。

首先創建您的項目目錄，然后將目錄更改為：

$ mkdir pycasts $ cd pycasts

現在您位于項目目錄中，您應該創建虛擬環境并激活它。使用任何讓您最高興的工具來執行此操作。此示例使用venv：

$ python3 -m venv .venv $ source .venv/bin/activate (.venv) $ python -m pip install --upgrade pip

現在您的環境已激活并pip升級，您需要安裝所需的依賴項以完成項目。您可以requirements.txt在本教程的可下載源代碼中找到一個文件：

獲取源代碼：?單擊此處獲取您將在本教程中使用 Django 和 Python 構建內容聚合器的源代碼。

打開source_code_setup/文件夾并安裝固定的依賴項。請務必將 替換為下載文件的實際路徑：

(.venv) $ python -m pip install -r

您現在應該已經安裝了 Django、feedparser、django-apscheduler和它們的子依賴項。

現在您擁有啟動和運行所需的所有工具，您可以設置 Django 并開始構建。要完成構建的這一步，您需要做以下四件事：

在當前工作目錄中創建您的 Django 項目，?/pycasts

創建一個podcastsDjango 應用

運行初始遷移

創建超級用戶

由于您已經熟悉 Django，因此您不會詳細探索這些步驟中的每一個。您可以繼續運行以下命令：

(.venv) $ django-admin startproject content_aggregator . (.venv) $ python manage.py startapp podcasts (.venv) $ python manage.py makemigrations && python manage.py migrate (.venv) $ python manage.py createsuperuser

如果事實證明您確實需要更深入地了解這些終端命令中的任何一個，您可以查看Django 第 1 部分入門。

按照 Django 的提示完成創建超級用戶帳戶后，在測試應用程序是否正常工作之前，您還需要進行一項更改。盡管應用程序在沒有它的情況下也能運行，但不要忘記將您的新podcasts應用程序添加到settings.py文件中：

# content_aggregator/settings.py # ... INSTALLED_APPS = [ "django.contrib.admin", "django.contrib.auth", "django.contrib.contenttypes", "django.contrib.sessions", "django.contrib.messages", "django.contrib.staticfiles", # My Apps "podcasts.apps.PodcastsConfig", ]

您將您的新應用INSTALLED_APPS列為"podcasts.apps.PodcastsConfig".

注意：如果您對為什么使用 verbosepodcasts.apps.PodcastsConfig而不是感到好奇podcasts，那么您可以在官方 Django 文檔中閱讀有關配置應用程序的更多信息。

TLDR；版本是，雖然使用的應用程序名稱，podcasts，應該工作的優良這個小應用程序，它被認為使用完整的最佳實踐AppConfig的名稱。

是時候嘗試一下您的新 Django 項目了。啟動 Django 服務器：

(.venv) $ python manage.py runserver

localhost:8000在瀏覽器中導航到，您應該會看到 Django 的默認成功頁面：

現在您已經設置了您的項目并且您有 Django 工作，請繼續下一步。

第 2 步：構建您的播客模型

此時，您應該已經設置了環境，安裝了依賴項，并且 Django 已成功運行。在此步驟結束時，您將定義和測試播客劇集的模型并將模型遷移到數據庫。

您的Episode模型不應僅反映您希望作為開發人員捕獲的信息。它還應該反映用戶希望看到的信息。跳入代碼并立即開始編寫模型很誘人，但這可能是一個錯誤。如果這樣做，您可能很快就會忘記用戶的觀點。畢竟，應用程序是為用戶設計的——甚至是像您或其他開發人員這樣的用戶。

在這一點上拿出筆和紙可能會很有用，但你應該做任何對你有用的事情。問問自己，“作為用戶，我想做什么？”?并一遍又一遍地回答這個問題，直到你窮盡所有的想法。然后，您可以通過思考作為開發人員的愿望，問問自己缺少什么。

在編寫數據庫模型時，這可能是一個很好的策略，它可以避免您以后需要添加額外的字段和運行不必要的遷移。

但是，如果您覺得缺少某個字段或屬性，請隨時擴展應用程序以在本教程末尾添加它。畢竟這是你的項目。讓它成為你自己的！

從用戶和開發人員的角度列出您的項目需求：

As a user, I would like to: - Know the title of an episode - Read a description of the episode - Know when an episode was published - Have a clickable URL so I can listen to the episode - See an image of the podcast so I can scroll to look for my favorite podcasts - See the podcast name As a developer, I would like to: - Have a uniquely identifiable attribute for each episode so I can avoid duplicating episodes in the database

您將在本教程的第 4 步中看到更多關于最后一點的內容。

根據您列出的要求Episode，您podcasts應用中的模型應如下所示：

# podcasts/models.py from django.db import models class Episode(models.Model): title = models.CharField(max_length=200) description = models.TextField() pub_date = models.DateTimeField() link = models.URLField() image = models.URLField() podcast_name = models.CharField(max_length=100) guid = models.CharField(max_length=50) def __str__(self) -> str: return f"{self.podcast_name}: {self.title}"

Django 最強大的部分之一是內置的管理區域。將劇集存儲在數據庫中是一回事，但您還希望能夠在管理區域與它們進行交互。您可以通過替換podcasts/admin.py文件中的代碼來告訴 Django 管理員您想要顯示您的劇集數據來做到這一點：

# podcasts/admin.py from django.contrib import admin from .models import Episode @admin.register(Episode) class EpisodeAdmin(admin.ModelAdmin): list_display = ("podcast_name", "title", "pub_date")

在將模型遷移到數據庫之前，您還需要做一件事。在 Django 3.2 中，您現在可以自定義自動創建的主鍵的類型。新的默認值與以前版本的 Django 中BigAutoField的Integer默認值相反。如果你現在運行遷移，你會看到這個錯誤：

(models.W042) Auto-created primary key used when not defining a primary key type, by default 'django.db.models.AutoField'. HINT: Configure the DEFAULT_AUTO_FIELD setting or the PodcastsConfig.default_auto_field attribute to point to a subclass of AutoField, e.g. 'django.db.models.BigAutoField'.

您可以通過向文件中的PodcastsConfig類添加額外的行來確保不會看到此錯誤app.py：

# podcasts/app.py from django.apps import AppConfig class PodcastsConfig(AppConfig): default_auto_field = "django.db.models.AutoField" name = "podcasts"

現在您的應用程序已配置為自動向所有模型添加主鍵。您還擁有數據應該是什么樣子的圖片，并將其表示在模型中。您現在可以運行Django 遷移以將您的Episode表包含在數據庫中：

(.venv) $ python manage.py makemigrations (.venv) $ python manage.py migrate

現在您已經遷移了更改，是時候測試它了！

本教程已經涵蓋了很多內容，因此為簡單起見，您將使用 Django 的內置測試框架進行單元測試。完成本教程中的項目后，如果您愿意，可以隨意使用pytest或其他測試框架重寫單元測試。

在您的podcasts/tests.py文件中，您可以添加：

# podcasts/tests.py from django.test import TestCase from django.utils import timezone from .models import Episode class PodCastsTests(TestCase): def setUp(self): self.episode = Episode.objects.create( title="My Awesome Podcast Episode", description="Look mom, I made it!", pub_date=timezone.now(), link="https://myawesomeshow.com", image="https://image.myawesomeshow.com", podcast_name="My Python Podcast", guid="de194720-7b4c-49e2-a05f-432436d3fetr", ) def test_episode_content(self): self.assertEqual(self.episode.description, "Look mom, I made it!") self.assertEqual(self.episode.link, "https://myawesomeshow.com") self.assertEqual( self.episode.guid, "de194720-7b4c-49e2-a05f-432436d3fetr" ) def test_episode_str_representation(self): self.assertEqual( str(self.episode), "My Python Podcast: My Awesome Podcast Episode" )

在上面的代碼中，您使用.setUp()來定義一個示例Episode對象。

您現在可以測試一些Episode屬性以確認模型按預期工作。從您的模型中測試字符串表示總是一個好主意，您在Episode.__str__().?字符串表示是您在調試代碼時將看到的內容，如果它準確地顯示您希望看到的信息，將使調試更容易。

現在您可以運行您的測試：

(.venv) $ python manage.py test

如果您的測試成功運行，恭喜！您現在為內容聚合器打下了良好的基礎，并且擁有了定義良好的數據模型。是時候進行第 3 步了。

第 3 步：創建您的主頁視圖

到現在為止，您應該有一個可運行的 Django 應用程序，其中包含您的Episode模型并通過了單元測試。在此步驟中，您將為主頁構建 HTML 模板，添加所需的CSS 和資產，將主頁添加到您的views.py文件中，并測試主頁是否正確呈現。

注意：編寫 HTML 和 CSS 超出了本教程的范圍，因此您不會涉及這些的原因和方法。但是，如果您對 HTML 或 CSS 有任何不理解或有疑問，可以在評論中聯系Real Python社區尋求見解。

在source_code_setup/您之前下載的文件夾中，您將找到一個名為static的文件夾和一個名為 的文件夾templates。您應該將這些文件夾復制到您的項目根文件夾pycasts/.?請務必將 替換為您在本地計算機上保存的實際路徑，并且不要忘記將點 (?.) 復制到當前工作目錄中：

(.venv) $ cp -r /static . (.venv) $ cp -r /templates .

現在您的項目根目錄中有 HTML 模板和靜態文件的文件夾，是時候將所有內容連接起來，以便 Django 知道它們存在。

轉到settings.py主content_aggregator應用程序中的文件。向下滾動直到到達該TEMPLATES部分，然后將templates/您之前創建的目錄添加到DIRS列表中。本教程使用 Django 3，它pathlib用于文件路徑：

# content_aggregator/settings.py # ... TEMPLATES = [ { "BACKEND": "django.template.backends.django.DjangoTemplates", "DIRS": [ BASE_DIR / "templates", ], "APP_DIRS": True, "OPTIONS": { "context_processors": [ "django.template.context_processors.debug", "django.template.context_processors.request", "django.contrib.auth.context_processors.auth", "django.contrib.messages.context_processors.messages", ], }, }, ]

您還需要將該static/文件夾添加到您的設置中。您可以通過向下滾動到文件STATIC部分settings.py并包含新創建static/文件夾的路徑來執行此操作：

# content_aggregator/settings.py # ... STATIC_URL = "/static/" STATICFILES_DIRS = [ BASE_DIR / "static", ]

Django 現在知道您的靜態資產和模板存在，但您還沒有完成。為了完成到目前為止您已經完成的工作，您還有一些任務需要在列表中打勾：

創建一個主頁視圖中views.py

創建 URL路徑

添加更多單元測試

您創建 URL 路徑和主頁視圖的順序并不重要。兩者都需要完成才能使應用程序正常工作，但您可以從列表的頂部開始并首先創建您的視圖類。

在您的podcasts應用程序中，打開您的views.py文件并將內容替換為以下代碼：

1# podcasts/views.py 2 3from django.views.generic import ListView 4 5from .models import Episode 6 7class HomePageView(ListView): 8 template_name = "homepage.html" 9 model = Episode 10 11 def get_context_data(self, **kwargs): 12 context = super().get_context_data(**kwargs) 13 context["episodes"] = Episode.objects.filter().order_by("-pub_date")[:10] 14 return context

您可能熟悉 Django 中基于函數的視圖，但 Django 也有內置的基于類的視圖。這些非常方便，可以減少您需要編寫的代碼量。

在上面的代碼片段中，您使用基于類的視圖將播客劇集發送到主頁：

第 7 行：您從ListView類繼承，以便您可以迭代劇集。默認情況下，它將迭代model = Episode第 9 行定義的所有劇集。

第 11 到 14 行：您覆蓋context數據并按最新的十集進行過濾，由發布日期確定pub_date。您想在此處進行過濾，因為否則可能會有數百（如果不是數千）劇集傳遞到主頁。

現在是時候為您的主頁提供一個 URL。您首先需要urls.py在您的podcasts應用程序中創建一個文件：

(.venv) $ touch podcasts/urls.py

現在您可以為HomePageView類添加路徑：

# podcasts/urls.py from django.urls import path from .views import HomePageView urlpatterns = [ path("", HomePageView.as_view(), name="homepage"), ]

在當前狀態下，應用程序仍然不會顯示您的主頁，因為主content_aggregator應用程序不知道podcasts/urls.py.?兩行代碼應該可以解決這個問題。在您的content_aggregator/urls.py文件中，添加突出顯示的代碼以將兩者連接在一起：

# podcasts/urls.py from django.contrib import admin from django.urls import path, include urlpatterns = [ path("admin/", admin.site.urls), path("", include("podcasts.urls")), ]

走到這一步做得很好！您現在應該能夠啟動您的應用程序并查看主頁。和以前一樣，啟動您的應用程序python manage.py runserver并前往localhost:8000：

您可以看到主頁有效，但沒有內容。即使沒有它，您仍然可以使用單元測試來測試內容是否能正確顯示。

在步驟 2 中，您為模型創建了單元測試。您還創建了.setUp()，它創建了一個Episode要測試的對象。您可以使用相同的測試集數據來測試您的主頁模板是否按預期工作。

除了測試主頁是否正確呈現劇集之外，最好同時測試是否使用了正確的模板以及導航到其 URL 是否返回了有效的 HTTP 狀態代碼。

對于這樣的單頁應用程序，它可能看起來有點矯枉過正——而且可能確實如此。但是，隨著任何應用程序的增長，您希望確保未來的更改不會破壞您的工作代碼。此外，如果您將此項目用作作品集，那么您應該表明您了解最佳實踐。

下面突出顯示的代碼是要添加到podcasts/tests.py文件中的新測試代碼：

# podcasts/tests.py from django.test import TestCase from django.utils import timezone from django.urls.base import reverse from datetime import datetime from .models import Episode class PodCastsTests(TestCase): def setUp(self): self.episode = Episode.objects.create( title="My Awesome Podcast Episode", description="Look mom, I made it!", pub_date=timezone.now(), link="https://myawesomeshow.com", image="https://image.myawesomeshow.com", podcast_name="My Python Podcast", guid="de194720-7b4c-49e2-a05f-432436d3fetr", ) def test_episode_content(self): self.assertEqual(self.episode.description, "Look mom, I made it!") self.assertEqual(self.episode.link, "https://myawesomeshow.com") self.assertEqual( self.episode.guid, "de194720-7b4c-49e2-a05f-432436d3fetr" ) def test_episode_str_representation(self): self.assertEqual( str(self.episode), "My Python Podcast: My Awesome Podcast Episode" ) def test_home_page_status_code(self): response = self.client.get("/") self.assertEqual(response.status_code, 200) def test_home_page_uses_correct_template(self): response = self.client.get(reverse("homepage")) self.assertTemplateUsed(response, "homepage.html") def test_homepage_list_contents(self): response = self.client.get(reverse("homepage")) self.assertContains(response, "My Awesome Podcast Episode")

和以前一樣，您可以使用python manage.py test.?如果此時您的所有測試都通過了，那么恭喜！

在這一步中，您成功創建了 HTML 模板和資產，構建了視圖類，并連接了所有 URL 路由。您還編寫了更多通過的單元測試。現在您已準備好進入下一步。

第 4 步：解析 Podcast RSS 提要

此時，您的應用程序應該看起來不錯！您擁有開始添加內容所需的一切。到這一步結束時，您應該可以輕松地使用 feedparser 庫來解析 RSS 提要并提取所需的數據。

在深入解析之前，RSS 提要究竟是什么？為什么要使用它來獲取播客數據？

首先，所有播客都有一個RSS 提要。這是播客應用程序獲取并向您顯示播客數據和劇集的基本方式。您通常可以在播客網站上找到提要 URL 鏈接。

此外，播客 RSS 提要需要看起來都一樣。這意味著當播客創建者將他們的提要提交到Apple Podcasts或Google Podcasts 等平臺時，提要必須遵守RSS 2.0 規范。

此要求有兩個方面的好處：

所有提要將具有相同的屬性，因此您可以重復使用代碼為任何給定的播客提取相同的數據，使您的代碼更易于維護和更干。

每個劇集都必須有一個guid分配給它，這使得該提要中的每個劇集都是獨一無二的。

您要解析的第一個提要是The Real Python Podcast提要。在瀏覽器中導航到https://realpython.com/podcasts/rpp/feed以查看提要的外觀。如果您覺得難以閱讀，可以安裝幾個瀏覽器插件之一來美化它。一個示例 Chrome 插件是XML Tree，但還有許多其他插件可用。

要使用 feedparser 解析提要，您可以使用parse()：

>>>

>>> import feedparser >>> feed = feedparser.parse("https://realpython.com/podcasts/rpp/feed")

parse()獲取提要并將其自動解析為可用的 Python 對象。然后，您可以使用標準點表示法訪問提要標簽，例如播客標題：

>>>

>>> podcast_title = feed.channel.title >>> podcast_title 'The Real Python Podcast'

您還可以使用括號表示法訪問標簽內的屬性：

>>>

>>> podcast_image = feed.channel.image["href"] >>> podcast_image 'https://files.realpython.com/media/real-python-logo-square.28474fda9228.png'

在使用 feedparser 解析的提要中，您還可以訪問稱為.entries.?這允許迭代提要中的每個元素。在您用播客劇集填充數據庫后，您將能夠使用.entries來檢查guid提要上每個播客劇集的 ，并檢查它是否存在于您的數據庫中。

注意：暫時不要實現以下代碼片段。只需閱讀它。當您創建 Django 自定義命令并將其用于您的項目時，您將在下一步中編寫類似的代碼。現在，只需瀏覽一下這段代碼，就可以了解如何使用 feedparser。

值得注意的是，您需要將來自 RSS 提要的發布日期轉換為一個datetime對象，以便將其保存到數據庫中。您將使用該dateutil庫來執行此操作：

# Example import feedparser from dateutil import parser from podcasts.models import Episode feed = feedparser.parse("https://realpython.com/podcasts/rpp/feed") podcast_title = feed.channel.title podcast_image = feed.channel.image["href"] for item in feed.entries: if not Episode.objects.filter(guid=item.guid).exists(): episode = Episode( title=item.title, description=item.description, pub_date=parser.parse(item.published), link=item.link, image=podcast_image, podcast_name=podcast_title, guid=item.guid, ) episode.save()

您尚未將此代碼放入文件的原因是您沒有在 Django 中運行它的好方法。既然您已經掌握了如何使用 feedparser，那么您將探索如何使用自定義命令來運行您的解析函數。

第 5 步：創建 Django 自定義命令

在最后一步中，您學習了如何使用 feedparser，但沒有以合理的方式運行與 Django ORM 交互的代碼。在這一步中，您將介紹如何使用自定義命令在您的項目中執行腳本，以便您可以在 Django 服務器或生產服務器也在運行時與其進行交互。

自定義命令利用該manage.py文件來運行您的代碼。當您運行 時manage.py，Django 將management/commands/目錄中的任何模塊注冊為可用命令。

注意：如果您想深入研究，請前往官方 Django 文檔以獲取有關自定義管理命令的更多信息。

首先創建適當的目錄和文件來存儲您的命令：

(.venv) $ mkdir -p podcasts/management/commands (.venv) $ touch podcasts/management/commands/startjobs.py

您幾乎可以為該文件命名任何您喜歡的名稱，但請注意，如果它以下劃線開頭，manage.py則不會注冊它。稍后在步驟 7 中，您將使用 django-apscheduler將作業添加到此文件中，這就是您將文件命名為startjobs.py.

為了測試您的設置，您將創建一個打印"It works!"到終端的基本命令。

每個命令都應該有一個Command類。該類需要一個.handle()方法，您可以將其視為類的主要方法。該.handle()方法包含您要執行的代碼：

# podcasts/management/commands/startjobs.py from django.core.management.base import BaseCommand class Command(BaseCommand): def handle(self, *args, **options): print("It works!")

現在從終端運行你的新命令：

(.venv) $ python manage.py startjobs

如果您看到It works!打印到終端，恭喜！您創建了第一個自定義命令。

現在是時候包含上一步中的 RSS 解析代碼，看看是否可以向數據庫中添加一些項目。繼續并更新您的startjobs.py代碼：

# podcasts/management/commands/startjobs.py from django.core.management.base import BaseCommand import feedparser from dateutil import parser from podcasts.models import Episode class Command(BaseCommand): def handle(self, *args, **options): feed = feedparser.parse("https://realpython.com/podcasts/rpp/feed") podcast_title = feed.channel.title podcast_image = feed.channel.image["href"] for item in feed.entries: if not Episode.objects.filter(guid=item.guid).exists(): episode = Episode( title=item.title, description=item.description, pub_date=parser.parse(item.published), link=item.link, image=podcast_image, podcast_name=podcast_title, guid=item.guid, ) episode.save()

這一次，當您運行自定義命令時，屏幕上不會打印任何內容，但您現在應該在主頁上顯示來自 The Real Python Podcast 的 Podcast 劇集。繼續嘗試。

你得到了什么？如果你還沒有導航到那里，現在去你的主頁：

你有沒有得到類似這張圖片的東西？如果是這樣，恭喜。有效。

既然您已經探索了如何使用自定義命令并設置并運行了第一個提要，您將在下一步中學習如何添加其他提要。

第 6 步：向 Python 內容聚合器添加其他提要

此時，您應該有一個可以工作的自定義命令來解析 The Real Python Podcast feed。在此步驟結束時，您將了解如何向自定義命令添加更多提要。

既然您有一個使用自定義命令成功解析的播客提要，您可能會想為每個提要一遍又一遍地重復相同的代碼。但是，這不是好的編碼習慣。您需要無需維護的 DRY 代碼。

您可能認為您可以遍歷一個提要 URL 列表并在每個項目上使用解析代碼，通常，這可以工作。然而，由于 django-apscheduler 的工作方式，這不是一個可行的解決方案。下一步將詳細介紹這一點。

相反，您需要重構您的代碼，以便為您需要解析的每個提要提供一個解析函數和一個單獨的函數。現在，您將分別調用這些方法。

注意：如本教程開頭所述，您目前只關注兩個提要。完成本教程并知道如何添加更多內容后，您可以通過選擇更多要添加的 RSS 提要來深入研究并自行練習。

與此同時，來自Talk Python to Me播客的Michael Kennedy非常友好地授予在本教程中使用他的播客提要的許可。謝謝你，邁克爾！

現在，您將開始探索代碼中的內容：

1# podcasts/management/commands/startjobs.py 2 3from django.core.management.base import BaseCommand 4 5import feedparser 6from dateutil import parser 7 8from podcasts.models import Episode 9 10def save_new_episodes(feed): 11 """Saves new episodes to the database. 12 13 Checks the episode GUID against the episodes currently stored in the 14 database. If not found, then a new `Episode` is added to the database. 15 16 Args: 17 feed: requires a feedparser object 18 """ 19 podcast_title = feed.channel.title 20 podcast_image = feed.channel.image["href"] 21 22 for item in feed.entries: 23 if not Episode.objects.filter(guid=item.guid).exists(): 24 episode = Episode( 25 title=item.title, 26 description=item.description, 27 pub_date=parser.parse(item.published), 28 link=item.link, 29 image=podcast_image, 30 podcast_name=podcast_title, 31 guid=item.guid, 32 ) 33 episode.save() 34 35def fetch_realpython_episodes(): 36 """Fetches new episodes from RSS for The Real Python Podcast.""" 37 _feed = feedparser.parse("https://realpython.com/podcasts/rpp/feed") 38 save_new_episodes(_feed) 39 40def fetch_talkpython_episodes(): 41 """Fetches new episodes from RSS for the Talk Python to Me Podcast.""" 42 _feed = feedparser.parse("https://talkpython.fm/episodes/rss") 43 save_new_episodes(_feed) 44 45class Command(BaseCommand): 46 def handle(self, *args, **options): 47 fetch_realpython_episodes() 48 fetch_talkpython_episodes()

正如剛才所討論的，您將解析代碼與各個提要分開以使其可重用。對于您添加的每個額外提要，您需要添加一個新的頂級函數。在此示例中，您已分別使用fetch_realpython_episodes()和對 The Real Python Podcast 和 Talk Python to Me Podcast 完成此操作fetch_talkpython_episodes()。

現在您知道如何向應用程序添加額外的提要，您可以繼續下一步，在那里您將了解如何自動運行自定義命令并定義運行它的計劃。

第 7 步：調度任務?django-apscheduler

此時，您應該有兩個或多個 RSS 提要排成一排，并準備好在每次運行新的自定義命令時進行解析。

在這最后一步中，您將：

設置django-apscheduler

為自定義命令添加時間表

將任務日志添加到您的應用程序

有機會在 Django 管理員中查看您的預定作業

django-apscheduler 包是 APScheduler 庫的 Django 實現。

注意：有關 APScheduler 和您可以使用的所有可能設置的詳細信息，請查看官方 APScheduler 文檔。您還可以在項目的GitHub存儲庫上閱讀有關 django-apscheduler 的更多信息。

您已經在虛擬環境中安裝了 django-apscheduler。要將其安裝到您的應用程序中，您還需要將其添加到INSTALLED_APPS您的settings.py文件中：

# content_aggregator/settings.py # ... INSTALLED_APPS = [ "django.contrib.admin", "django.contrib.auth", "django.contrib.contenttypes", "django.contrib.sessions", "django.contrib.messages", "django.contrib.staticfiles", # My Apps "podcasts.apps.PodcastsConfig", # Third Party Apps "django_apscheduler", ]

要創建 django-apscheduler 模型，您需要運行數據庫遷移命令：

(.venv) $ python manage.py migrate

此命令應用 django-apscheduler 正常工作所需的數據庫遷移。

注意：不需要先運行makemigrations，因為 django-apscheduler 包包含它自己的遷移文件。

現在 django-apscheduler 已安裝到您的應用程序中，您將簡要了解它的工作原理。更詳細的解釋請查看官方文檔。

您要在自定義命令中運行的每個任務都稱為作業。您的應用程序中共有三個作業：一個用于您希望解析的每個播客提要，第三個用于從數據庫中刪除舊作業。

django-apscheduler 包將您的作業存儲在數據庫中，它還將存儲所有成功和不成功的作業運行。擁有此歷史記錄對您作為開發人員或站點管理員來說非常有用，因為您可以監控任何任務是否失敗。但是如果不定期從數據庫中清除這些，您的數據庫將很快填滿，因此從數據庫中清除舊歷史記錄是一種很好的做法。這也將按計劃進行。

即使作業歷史記錄將存儲在數據庫中，如果發生任何錯誤，最好將它們記錄下來以進行調試。您可以通過將此代碼添加到您的應用程序來添加一些基本的日志記錄設置settings.py：

# content_aggregator/settings.py # ... LOGGING = { "version": 1, "disable_existing_loggers": False, "handlers": { "console": { "class": "logging.StreamHandler", }, }, "root": { "handlers": ["console"], "level": "INFO", }, }

現在您已經添加了日志記錄設置，您需要在startjobs.py文件中實例化它。您現在將包含一些導入startjobs.py，稍后您將詳細介紹這些導入。添加調度程序所需的日志記錄和其他一些導入語句：

1# podcasts/management/commands/startjobs.py 2 3# Standard Library 4import logging 5 6# Django 7from django.conf import settings 8from django.core.management.base import BaseCommand 9 10# Third Party 11import feedparser 12from dateutil import parser 13from apscheduler.schedulers.blocking import BlockingScheduler 14from apscheduler.triggers.cron import CronTrigger 15from django_apscheduler.jobstores import DjangoJobStore 16from django_apscheduler.models import DjangoJobExecution 17 18# Models 19from podcasts.models import Episode 20 21logger = logging.getLogger(__name__)

這可能是一次添加的大量導入語句，所以讓我們挑選出需要一些解釋的類：

第 13 行：?BlockingScheduler是將運行您的作業的調度程序。它是阻塞的，所以它將是進程上唯一運行的東西。

第 14 行：?CronTrigger是您將用于調度的觸發器類型。

第15行：?DjangoJobStore將決定如何將作業存儲。在這種情況下，您希望它們在數據庫中。

第 16 行：您將用于DjangoJobExecution運行剛才提到的清理函數。

接下來，您需要使用所有導入語句并設置調度程序、觸發器和作業存儲。

您已經編寫了三個工作職能中的兩個，所以現在是添加第三個的時候了。在您的Command班級上方，添加您的工作職能：

# podcasts/management/commands/startjobs.py # ... def delete_old_job_executions(max_age=604_800): """Deletes all apscheduler job execution logs older than `max_age`.""" DjangoJobExecution.objects.delete_old_job_executions(max_age)

該max_age參數是表示為整數的秒數。請注意，604,800 秒等于 1 周。

下一步是在自定義命令的.handle()函數中創建您的作業存儲和調度程序實例。您還可以在第一份工作中添加：

1# podcasts/management/commands/startjobs.py 2 3# ... 4 5def handle(self, *args, **options): 6 scheduler = BlockingScheduler(timezone=settings.TIME_ZONE) 7 scheduler.add_jobstore(DjangoJobStore(), "default") 8 9 scheduler.add_job( 10 fetch_realpython_episodes, 11 trigger="interval", 12 minutes=2, 13 id="The Real Python Podcast", 14 max_instances=1, 15 replace_existing=True, 16 ) 17 logger.info("Added job: The Real Python Podcast.")

您將在上面看到您已成功創建scheduler實例并添加了作業存儲。然后，您創建了您的第一份工作 — The Real Python Podcast。

該.add_job()方法需要一些參數才能成功創建作業：

第 10 行：第一個參數需要一個函數，因此您將fetch_realpython_episodes之前創建的函數對象傳遞給它。請注意它在傳遞時如何沒有調用括號。

第 11 和 12 行：您必須設置觸發器。在本教程中，您將執行之間的間隔設置為兩分鐘。這只是為了讓您可以測試并查看它是否適合自己。但是，絕對不應該在生產環境中頻繁使用它。您還可以傳遞seconds和hours作為參數，因此如果您在實時環境中托管此應用程序，您可以設置更現實的更新間隔。

第 13 行：所有作業都必須有一個 ID。Django 管理員也將使用該 ID，因此請選擇一個可讀且有意義的名稱。

第15行：將replace_existing現有的工作和防止重復，當你重新啟動應用程序關鍵字參數內容替換。

您可以查看官方 APScheduler 文檔以獲取所有可接受參數的完整列表.add_job()。

現在，你有你的第一份工作計劃，你可以繼續前進，最后兩個職位增加.handle()，并在電話中添加scheduler.start()和scheduler.shutdown()。哦，讓我們也加入一些日志記錄。

您的自定義命令類現在應如下所示：

# podcasts/management/commands/startjobs.py # ... class Command(BaseCommand): help = "Runs apscheduler." def handle(self, *args, **options): scheduler = BlockingScheduler(timezone=settings.TIME_ZONE) scheduler.add_jobstore(DjangoJobStore(), "default") scheduler.add_job( fetch_realpython_episodes, trigger="interval", minutes=2, id="The Real Python Podcast", max_instances=1, replace_existing=True, ) logger.info("Added job: The Real Python Podcast.") scheduler.add_job( fetch_talkpython_episodes, trigger="interval", minutes=2, id="Talk Python Feed", max_instances=1, replace_existing=True, ) logger.info("Added job: Talk Python Feed.") scheduler.add_job( delete_old_job_executions, trigger=CronTrigger( day_of_week="mon", hour="00", minute="00" ), # Midnight on Monday, before start of the next work week. id="Delete Old Job Executions", max_instances=1, replace_existing=True, ) logger.info("Added weekly job: Delete Old Job Executions.") try: logger.info("Starting scheduler...") scheduler.start() except KeyboardInterrupt: logger.info("Stopping scheduler...") scheduler.shutdown() logger.info("Scheduler shut down successfully!")

您可能還記得之前的內容，該調度程序使用BlockingScheduler.?在一個終端中，您可以像以前一樣使用python manage.py startjobs.?在單獨的終端進程中，啟動 Django 服務器。當您查看管理儀表板時，您現在會看到您的作業已注冊，并且您可以查看歷史記錄：

在這最后一步中發生了很多事情，但您已經完成了一個功能正常的應用程序！您已經成功地了解了如何使用 django-apscheduler 按照定義的時間表自動運行您的自定義命令。不小的壯舉。做得好！

從頭開始構建諸如內容聚合器之類的項目從來都不是一項快速或簡單的任務，您應該為自己成功地做到這一點而感到自豪。做一些新的和推動你一點的事情只會幫助你成長為一名開發人員，無論你的資歷有多高。

結論

您在這個基于項目的教程中涵蓋了很多內容。很好！

在本教程中，您學習了：

如何使用 feedparser處理 RSS 提要

如何創建和使用自定義管理命令

如何使用 django-apscheduler 根據個性化計劃自動執行自定義命令

如何向Django 應用程序添加基本單元測試

如果您還沒有這樣做，請單擊下面的鏈接下載本教程的代碼，以便您可以使用 Python 構建自己的內容聚合器：

下一步

您可以通過多種方式自定義和更改此應用程序，以表達自己作為開發人員的身份，尤其是當您計劃將其用作作品集時。將這個項目提升到一個新的水平將幫助你在未來的工作申請中脫穎而出。

這里有一些想法可以將您的項目提升到一個新的水平：

添加更多提要！對數據科學感興趣？查看數據科學播客的終極列表。

更改內容類型。如果播客不是你的東西，也許足球新聞是？或者，您可能更喜歡聽金融播客？無論您有什么興趣，都可以使用這個項目作為墊腳石，為您的激情之一創建聚合器。

添加用戶帳戶，以便用戶可以訂閱他們感興趣的提要。然后在一天結束時，向他們發送一封電子郵件，其中包含他們訂閱的新內容。查看有關Django 視圖授權的教程以獲得這方面的幫助。

is_published向Episode模型添加布爾標志，以便管理員可以手動策劃和確認主頁上顯示的劇集。

featured向Episode模型添加一個字段，以便您可以在主頁上突出顯示選定的播客。

重新設計應用程序以使其成為您自己的應用程序。自定義 CSS 或刪除 Bootstrap 4 以支持Tailwind CSS。世界是你的牡蠣。

將應用程序部署到生產環境——例如，通過在 Heroku 上托管您的 Django 應用程序。

自定義 Django 管理員并增強您的 Django 管理員體驗。

Django Python

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。