StaticPageNamespace.fetch_upstream() to import markdown pages from upstream_url

This allows to pull externally managed pages into a (protected) wiki namespace.

src/core/models/pages.py

-from typing import Optional, Tuple
+import logging
+from typing import Dict, Optional, Tuple
 import uuid
 
 from django.conf import settings
@@ -8,6 +9,7 @@ from django.contrib.postgres.fields import ArrayField
 from django.core.exceptions import ValidationError
 from django.db import models
 from django.db.models import Q
+from django.utils import timezone
 from django.utils.functional import cached_property
 from django.utils.html import strip_tags
 from django.utils.translation import gettext_lazy as _
@@ -17,6 +19,7 @@ from core.markdown import render_markdown_ex, store_relationships
 from ..fields import ConferenceReference
 from .conference import Conference
 from .users import PlatformUser
+from ..utils import GitRepo
 
 
 class StaticPageNamespace(models.Model):
@@ -57,6 +60,80 @@ class StaticPageNamespace(models.Model):
     def __str__(self):
         return self.prefix
 
+    @staticmethod
+    def _extract_doc_metadata(markup) -> Tuple[Dict[str, str], str]:
+        if markup.startswith('---\n'):
+            metadata_raw, content = markup[4:].split('---\n', maxsplit=1)
+            metadata = {}
+            for line in metadata_raw.splitlines():
+                k, v = line.split(":", maxsplit=1)
+                metadata[k.strip()] = v.strip()
+        else:
+            metadata = {}
+            content = markup
+
+        # TODO: optionally convert mkdocs-specific non-markdown syntax?
+
+        return metadata, content
+
+    def fetch_upstream(self):
+        if not (url := self.upstream_url):
+            return {}
+
+        with GitRepo(url) as repo:
+            docs = repo.get_documents(glob="*.md")
+
+        existing_pages = {page.slug: page for page in StaticPage.objects.filter(conference=self.conference, slug__istartswith=self.prefix)}
+        obsolete_pages = set(existing_pages.keys())
+
+        for doc_filename, document_raw in docs.items():
+            page_slug = self.prefix.lower() + doc_filename.rstrip(".md")
+            page = existing_pages.get(page_slug)  # type: StaticPage
+
+            try:
+                doc_metadata, document = self._extract_doc_metadata(document_raw)
+
+                if page is not None:
+                    # will update the existing page, note that we've seen it
+                    obsolete_pages.remove(page_slug)
+                    current_rev = page.revisions.get(revision=page.public_revision)
+                    if current_rev.body != document:
+                        # the document content has been changed
+                        page_rev = page.revisions.create(
+                            title=doc_metadata.get('title') or current_rev.title,
+                            body=document,
+                            is_draft=False,
+                            author=None,
+                            timestamp=timezone.now(),
+                        )
+                        page_rev.set_public()
+                        logging.info(f'wiki page "{page_slug}" updated (old revision #{current_rev.revision}, new #{page_rev.revision})')
+                        current_rev.delete()
+                    else:
+                        logging.debug(f'wiki page "{page_slug}" unchanged (revision #{current_rev.revision})')
+
+                else:
+                    # create new page
+                    page = StaticPage.objects.create(conference=self.conference, slug=page_slug)
+                    page_rev = page.revisions.create(
+                        title=doc_metadata.get('title') or page_slug[len(self.prefix):].capitalize(),
+                        body=document,
+                        is_draft=False,
+                        author=None,
+                        timestamp=timezone.now(),
+                    )
+                    page_rev.set_public()
+                    logging.info(f'wiki page "{page_slug}" created (revision #{page_rev.revision})')
+            except Exception:
+                logging.exception(f'wiki page "{page_slug}" failed to import')
+                if page is not None:
+                    # don't remove the previous version if importing a new one fails
+                    obsolete_pages.remove(page_slug)
+
+        for page_slug in obsolete_pages:
+            existing_pages[page_slug].delete()
+            logging.info(f'wiki page "{page_slug}" deleted (removed upstream)')
+
 
 class StaticPageManager(models.Manager):
     def accessible_by_user(self, user: PlatformUser, conference: Conference, language: Optional[str] = ''):