add remote url page type

shiftregister/pages/admin.py

@@ -23,6 +23,18 @@ def reimport(modeladmin, request, queryset):
         )
 
 
+def fetch_remote_content(modeladmin, request, queryset):
+    remote_pages = queryset.filter(kind=Page.REMOTE_CONTENT)
+    for page in remote_pages:
+        page.fetch_remote_content()
+
+    modeladmin.message_user(
+        request,
+        f"{remote_pages.count()} Remote-Inhalte wurden aktualisiert.",
+        messages.SUCCESS,
+    )
+
+
 @admin.register(Page)
 class PageAdmin(admin.ModelAdmin):
     fields = (
@@ -33,6 +45,8 @@ class PageAdmin(admin.ModelAdmin):
         "kind",
         "show_in_footer_nav",
         "show_in_main_nav",
+        "remote_url",
+        "import_error",
     )
     list_display = (
         "url",
@@ -41,5 +55,7 @@ class PageAdmin(admin.ModelAdmin):
         "kind",
         "show_in_footer_nav",
         "show_in_main_nav",
+        "remote_url",
+        "import_error",
     )
-    actions = (reimport,)
+    actions = (reimport, fetch_remote_content)

shiftregister/pages/migrations/0004_page_import_error_page_remote_url_alter_page_kind.py

@@ -0,0 +1,36 @@
+# Generated by Django 5.0.4 on 2025-05-21 23:20
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("pages", "0003_page_show_in_footer_nav_page_show_in_main_nav"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="page",
+            name="import_error",
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AddField(
+            model_name="page",
+            name="remote_url",
+            field=models.URLField(blank=True, null=True),
+        ),
+        migrations.AlterField(
+            model_name="page",
+            name="kind",
+            field=models.CharField(
+                choices=[
+                    ("redirect", "Redirect"),
+                    ("regular", "Regular page"),
+                    ("remote", "Remote content"),
+                ],
+                default="regular",
+                max_length=8,
+            ),
+        ),
+    ]

shiftregister/pages/models.py

@@ -1,3 +1,5 @@
+import requests
+from bs4 import BeautifulSoup
 from django.db import models
 
 # Create your models here.
@@ -6,9 +8,11 @@ from django.db import models
 class Page(models.Model):
     REDIRECT = "redirect"
     REGULAR = "regular"
+    REMOTE_CONTENT = "remote"
     KIND_CHOICES = [
         (REDIRECT, "Redirect"),
         (REGULAR, "Regular page"),
+        (REMOTE_CONTENT, "Remote content"),
     ]
 
     url = models.fields.SlugField(unique=True)
@@ -19,9 +23,32 @@ class Page(models.Model):
     show_in_footer_nav = models.BooleanField(default=True)
     show_in_main_nav = models.BooleanField(default=False)
 
+    # Fields for remote content
+    remote_url = models.URLField(blank=True, null=True)
+    import_error = models.BooleanField(default=False)
+
     def __str__(self):
-        return (
+        if self.kind == Page.REDIRECT:
-            f"{self.get_kind_display()} {self.url}" + f" => {self.content}"
+            return f"{self.get_kind_display()} {self.url} => {self.content}"
-            if self.kind == Page.REDIRECT
+        elif self.kind == Page.REMOTE_CONTENT:
-            else ""
+            return f"{self.get_kind_display()} {self.url} from {self.remote_url}"
-        )
+        return f"{self.get_kind_display()} {self.url}"
+
+    def fetch_remote_content(self):
+        if self.kind != self.REMOTE_CONTENT or not self.remote_url:
+            return
+
+        try:
+            response = requests.get(self.remote_url)
+            response.raise_for_status()
+
+            soup = BeautifulSoup(response.text, "html.parser")
+            body_content = soup.find("body")
+
+            if body_content:
+                self.content = f"<div class='content'>{str(body_content)}</div>"
+                self.import_error = False
+        except (requests.RequestException, Exception):
+            self.import_error = True
+
+        self.save()

shiftregister/pages/tasks.py

@@ -0,0 +1,10 @@
+from celery import shared_task
+
+from .models import Page
+
+
+@shared_task
+def fetch_all_remote_content():
+    remote_pages = Page.objects.filter(kind=Page.REMOTE_CONTENT)
+    for page in remote_pages:
+        page.fetch_remote_content()

shiftregister/pages/tests.py

@@ -1,3 +1,104 @@
+from unittest.mock import Mock, patch
+
 from django.test import TestCase
+from django.urls import reverse
+
+from .models import Page
+from .tasks import fetch_all_remote_content
 
 # Create your tests here.
+
+
+class RemoteContentTests(TestCase):
+    def setUp(self):
+        self.remote_page = Page.objects.create(
+            url="test-remote",
+            title="Test Remote",
+            kind=Page.REMOTE_CONTENT,
+            remote_url="https://example.com/test",
+            content="",
+        )
+        self.regular_page = Page.objects.create(
+            url="test-regular",
+            title="Test Regular",
+            kind=Page.REGULAR,
+            content="Regular content",
+        )
+
+    @patch("requests.get")
+    def test_fetch_remote_content_success(self, mock_get):
+        # Mock successful response
+        mock_response = Mock()
+        mock_response.text = """
+        <html>
+            <body>
+                <h1>Test Content</h1>
+                <p>Some test content</p>
+            </body>
+        </html>
+        """
+        mock_get.return_value = mock_response
+
+        # Fetch content
+        self.remote_page.fetch_remote_content()
+
+        # Verify the page was updated
+        self.remote_page.refresh_from_db()
+        self.assertIn("<h1>Test Content</h1>", self.remote_page.content)
+        self.assertIn("<p>Some test content</p>", self.remote_page.content)
+        self.assertFalse(self.remote_page.import_error)
+        self.assertTrue(self.remote_page.content.startswith("<div class='content'>"))
+
+    @patch("requests.get")
+    def test_fetch_remote_content_error(self, mock_get):
+        # Mock failed request
+        mock_get.side_effect = Exception("Connection error")
+
+        # Fetch content
+        self.remote_page.fetch_remote_content()
+
+        # Verify error state
+        self.remote_page.refresh_from_db()
+        self.assertTrue(self.remote_page.import_error)
+        self.assertEqual(self.remote_page.content, "")
+
+    def test_fetch_regular_page(self):
+        # Try to fetch content for a regular page
+        self.regular_page.fetch_remote_content()
+
+        # Verify nothing changed
+        self.regular_page.refresh_from_db()
+        self.assertEqual(self.regular_page.content, "Regular content")
+        self.assertFalse(self.regular_page.import_error)
+
+    @patch("requests.get")
+    def test_fetch_all_remote_content(self, mock_get):
+        # Create another remote page
+        Page.objects.create(
+            url="test-remote-2",
+            title="Test Remote 2",
+            kind=Page.REMOTE_CONTENT,
+            remote_url="https://example.com/test2",
+            content="",
+        )
+
+        # Mock successful response
+        mock_response = Mock()
+        mock_response.text = "<html><body>Test content</body></html>"
+        mock_get.return_value = mock_response
+
+        # Run the task
+        fetch_all_remote_content()
+
+        # Verify both remote pages were updated
+        remote_pages = Page.objects.filter(kind=Page.REMOTE_CONTENT)
+        for page in remote_pages:
+            self.assertIn("Test content", page.content)
+            self.assertFalse(page.import_error)
+
+    def test_page_view_remote_content(self):
+        # Test the view with remote content
+        url = reverse("pages:view", kwargs={"slug": self.remote_page.url})
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        self.assertTemplateUsed(response, "page.html")

shiftregister/settings.py

@@ -200,6 +200,12 @@ CELERY_BEAT_SCHEDULE = {
         "task": "shiftregister.messaging.tasks.fetch_messages",
         "schedule": env.float("MESSAGE_FETCH_INTERVAL", default=300.0),  # seconds
     },
+    "fetch-remote-content-every-600-seconds": {
+        "task": "shiftregister.pages.tasks.fetch_all_remote_content",
+        "schedule": env.float(
+            "REMOTE_CONTENT_FETCH_INTERVAL", default=600.0
+        ),  # seconds
+    },
 }
 
 CELERY_BEAT_SCHEDULE_FILENAME = str(BASE_DIR / "storage" / "celerybeat-schedule")