Migrating from Mezzanine content management system to static page generation

#!/usr/bin/env python3
"""One-time import of the Mezzanine fixture and local media into Hugo."""

from __future__ import annotations

import json
import re
import shutil
from collections import defaultdict
from datetime import datetime
from pathlib import Path, PurePosixPath
from textwrap import dedent
from typing import Any

from bs4 import BeautifulSoup
from markdownify import markdownify


ROOT = Path(__file__).resolve().parents[1]
FIXTURE = ROOT / "01_export/dump_431_excerpt-202610.json"
SOURCE_MEDIA = ROOT / "01_export/blog/media"
SITE = ROOT / "hugo"
CONTENT = SITE / "content"
PUBLISHED_STATUS = 2  # Confirmed against the live site's published posts.
AUTHOR_NAME = "Lars Steinke"  # Public display name shown by the live site.


def toml_value(value: Any) -> str:
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, str):
        return json.dumps(value, ensure_ascii=False)
    if isinstance(value, (int, float)):
        return str(value)
    if isinstance(value, list):
        return "[" + ", ".join(toml_value(item) for item in value) + "]"
    raise TypeError(f"Unsupported TOML value: {type(value).__name__}")


def write_content(path: Path, fields: dict[str, Any], body: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    front_matter = "+++\n" + "".join(
        f"{key} = {toml_value(value)}\n" for key, value in fields.items() if value is not None
    )
    path.write_text(front_matter + "+++\n" + body.strip() + "\n", encoding="utf-8")


def normalize_html(body: str) -> str:
    body = (
        body.replace("/static/media/uploads/", "/media/uploads/")
        .replace("/static/img/", "/img/")
    )
    gpg_form = re.compile(
        r'<form\b(?=[^>]*action=["\']https://pgp\.cs\.uu\.nl/mk_path\.cgi["\'])[^>]*>.*?</form>',
        re.IGNORECASE | re.DOTALL,
    )
    return gpg_form.sub("{{< gpg-trust-path >}}", body)


def normalize_blog_links(body: str, post_slugs: set[str]) -> str:
    link = re.compile(r'''(?P<quote>["'])(?P<path>/blog/[^"'?#]+)(?P<tail>[?#][^"']*)?(?P=quote)''')

    def add_trailing_slash(match: re.Match[str]) -> str:
        path = match.group("path")
        slug = path.rstrip("/").rsplit("/", 1)[-1]
        if slug in post_slugs and not path.endswith("/"):
            path += "/"
        return f'{match.group("quote")}{path}{match.group("tail") or ""}{match.group("quote")}'

    return link.sub(add_trailing_slash, body)


def promote_multiline_code(body: str) -> str:
    soup = BeautifulSoup(body, "html.parser")
    for code in list(soup.find_all("code")):
        if code.find("br"):
            code_text = dedent(code.get_text(separator="\n", strip=False)).strip()
            if len([line for line in code_text.splitlines() if line.strip()]) < 2:
                continue
            pre = soup.new_tag("pre")
            block = soup.new_tag("code")
            block.string = code_text
            pre.append(block)
            code.replace_with(pre)
    return str(soup)


def convert_html_to_markdown(body: str, post_slugs: set[str]) -> str:
    prepared = promote_multiline_code(body)
    normalized = normalize_blog_links(normalize_html(prepared), post_slugs)
    return markdownify(normalized, heading_style="ATX", bullets="-").strip()


def date_value(value: str | None) -> str | None:
    if not value:
        return None
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    return parsed.isoformat()


def copy_media() -> None:
    for source, target in (
        (SOURCE_MEDIA / "img", SITE / "static/img"),
        (SOURCE_MEDIA / "uploads", SITE / "static/media/uploads"),
    ):
        if not source.is_dir():
            if target.is_dir():
                continue  # Tracked copies are sufficient when ignored source media is absent.
            raise FileNotFoundError(f"Expected source or copied media directory: {source}")
        for item in source.rglob("*"):
            if item.is_file():
                relative = item.relative_to(source)
                destination = target / relative
                destination.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(item, destination)

    for required in (SITE / "static/img/lsbw.svg", SITE / "static/img/favicon.ico"):
        if not required.is_file():
            raise FileNotFoundError(f"Missing required site asset: {required}")


def load_records() -> list[dict[str, Any]]:
    with FIXTURE.open(encoding="utf-8") as source:
        rows = json.load(source)
    if not isinstance(rows, list):
        raise ValueError(f"Expected a Django fixture list in {FIXTURE}")
    return rows


def convert() -> None:
    rows = load_records()
    copy_media()
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in rows:
        grouped[row["model"]].append(row)

    categories = {
        row["pk"]: row["fields"] for row in grouped["blog.blogcategory"]
    }
    keywords = {
        row["pk"]: row["fields"] for row in grouped["generic.keyword"]
    }
    category_by_slug = {value["slug"]: value for value in categories.values()}
    tag_title_by_slug = {value["slug"]: value["title"] for value in keywords.values()}
    for category in categories.values():
        tag_title_by_slug.setdefault(category["slug"], category["title"])

    keyword_by_post: dict[str, list[str]] = defaultdict(list)
    for row in grouped["generic.assignedkeyword"]:
        fields = row["fields"]
        content_type = fields.get("content_type")
        if content_type == ["blog", "blogpost"]:
            keyword = keywords.get(fields.get("keyword"))
            if keyword:
                keyword_by_post[str(fields.get("object_pk"))].append(keyword["slug"])

    post_rows = sorted(
        (row for row in grouped["blog.blogpost"] if row["fields"].get("status") == PUBLISHED_STATUS),
        key=lambda row: row["fields"].get("publish_date") or "",
        reverse=True,
    )
    posts_by_pk = {str(row["pk"]): row for row in post_rows}
    published_slugs = {row["fields"]["slug"] for row in post_rows}
    post_slugs: set[str] = set()
    archive_months: set[tuple[int, int]] = set()

    (CONTENT / "blog").mkdir(parents=True, exist_ok=True)
    write_content(CONTENT / "_index.md", {"title": "Blog", "type": "home"}, "")
    write_content(CONTENT / "blog" / "_index.md", {"title": "Blog", "type": "blog"}, "")

    for row in post_rows:
        fields = row["fields"]
        slug = fields.get("slug", "")
        if not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", slug):
            raise ValueError(f"Unexpected blog slug in fixture: {slug!r}")
        if slug in post_slugs:
            raise ValueError(f"Duplicate blog slug in fixture: {slug}")
        post_slugs.add(slug)

        post_tag_slugs: list[str] = []
        for category_pk in fields.get("categories", []):
            category = categories.get(category_pk)
            if not category:
                raise ValueError(f"Missing blog category {category_pk} for post {slug}")
            post_tag_slugs.append(category["slug"])

        post_tag_slugs.extend(keyword_by_post.get(str(row["pk"]), []))
        post_tag_slugs = list(dict.fromkeys(post_tag_slugs))
        for tag_slug in post_tag_slugs:
            if tag_slug not in tag_title_by_slug:
                raise ValueError(f"Missing tag term {tag_slug!r} for post {slug}")
        tags = [tag_title_by_slug[tag_slug] for tag_slug in post_tag_slugs]

        related_slugs = []
        for related_pk in fields.get("related_posts", []):
            related = posts_by_pk.get(str(related_pk))
            if related:
                related_slugs.append(related["fields"]["slug"])

        published = date_value(fields.get("publish_date"))
        if not published:
            raise ValueError(f"Published post {slug} has no publish date")
        published_date = datetime.fromisoformat(published)
        archive_months.add((published_date.year, published_date.month))
        front_matter = {
            "title": fields["title"],
            "slug": slug,
            "date": published,
            "lastmod": date_value(fields.get("updated")),
            "draft": False,
            "type": "blog",
            "description": fields.get("description", ""),
            "tags": tags,
            "author": [AUTHOR_NAME],
            "related_posts": related_slugs,
        }
        write_content(
            CONTENT / "blog" / f"{slug}.md",
            front_matter,
            convert_html_to_markdown(fields.get("content", ""), published_slugs),
        )
        (CONTENT / "blog" / f"{slug}.html").unlink(missing_ok=True)

    # Categories become tags; remove generated category pages and redirect their old routes.
    category_slugs = set(category_by_slug)
    for slug in category_slugs:
        (CONTENT / "blog/category" / slug / "_index.md").unlink(missing_ok=True)

    # Hugo taxonomies generate tag term pages dynamically; remove old imported term files.
    for slug in tag_title_by_slug:
        (CONTENT / "blog/tag" / slug / "_index.md").unlink(missing_ok=True)

    (CONTENT / "blog/author/lss" / "_index.md").unlink(missing_ok=True)

    for year, month in archive_months:
        (CONTENT / "blog/archive" / str(year) / str(month) / "_index.md").unlink(missing_ok=True)
    write_content(
        CONTENT / "blog/archive" / "_index.md",
        {"title": "Archive", "type": "archive-index", "draft": False},
        "",
    )

    # Convert the Mezzanine page tree. The dynamic contact form is intentionally
    # omitted; keep its route as a landing page for the static contact details.
    richtext = {row["pk"]: row["fields"] for row in grouped["pages.richtextpage"]}
    gallery = {row["pk"]: row["fields"] for row in grouped["galleries.gallery"]}
    gallery_images = sorted(
        grouped["galleries.galleryimage"], key=lambda row: row.get("pk", 0)
    )
    for row in grouped["pages.page"]:
        fields = row["fields"]
        if fields.get("status") != PUBLISHED_STATUS:
            continue
        slug = fields["slug"].strip("/")
        if slug == "contact":
            body = (
                "See [About](/contact/about/) for profile and contact links, or "
                "[Legal info](/contact/legal/) for the site's legal details."
            )
            path = CONTENT / "contact" / "_index.md"
        elif fields.get("content_model") == "gallery":
            gallery_fields = gallery.get(row["pk"], {})
            body = convert_html_to_markdown(gallery_fields.get("content", ""), published_slugs)
            body += "\n\n{{< gallery >}}"
            gallery_items = []
            for image_row in gallery_images:
                image = image_row["fields"]
                if image.get("gallery") != row["pk"]:
                    continue
                relative = PurePosixPath(image["file"])
                if relative.is_absolute() or ".." in relative.parts:
                    raise ValueError(f"Unsafe gallery asset path: {image['file']!r}")
                asset = SITE / "static/media" / relative
                if not asset.is_file():
                    raise FileNotFoundError(f"Missing gallery image: {asset}")
                url = "/media/" + relative.as_posix()
                gallery_items.append({"url": url, "caption": image.get("description", "")})
            gallery_data_path = SITE / "data/gallery.json"
            gallery_data_path.parent.mkdir(parents=True, exist_ok=True)
            gallery_data_path.write_text(
                json.dumps({"images": gallery_items}, ensure_ascii=False, indent=2) + "\n",
                encoding="utf-8",
            )
            path = CONTENT / f"{slug}.md"
        else:
            page_content = richtext.get(row["pk"], {}).get("content", "")
            body = convert_html_to_markdown(page_content, published_slugs)
            path = CONTENT / f"{slug}.md"

        if slug != "contact":
            (CONTENT / f"{slug}.html").unlink(missing_ok=True)

        page_fields = {
            "title": fields["title"],
            "type": "gallery" if fields.get("content_model") == "gallery" else "pages",
            "date": date_value(fields.get("publish_date")),
            "description": (
                "Contact information and legal details."
                if slug == "contact"
                else fields.get("description", "")
            ),
            "draft": False,
        }
        write_content(path, page_fields, body)

    data_dir = SITE / "data"
    data_dir.mkdir(parents=True, exist_ok=True)
    (data_dir / "sidebar.json").unlink(missing_ok=True)

    print(
        f"Converted {len(post_rows)} published posts, "
        f"{sum(fields.get('status') == PUBLISHED_STATUS for fields in (r['fields'] for r in grouped['pages.page']))} pages, "
        f"and {sum(1 for r in gallery_images)} gallery images into {SITE}"
    )


if __name__ == "__main__":
    convert()