"""One-shot WordPress WXR -> Hugo importer.
Run once during the migration:
python tools/fetch_media.py # pull the binaries off the live host
python tools/build_data.py # termine / members / memoriam
python tools/wp2hugo.py # content bundles + assets
Afterwards the generated Markdown in content/ is the source of truth and this
script is kept only so the import is reproducible and reviewable.
"""
from __future__ import annotations
import json
import re
import shutil
import sys
from dataclasses import dataclass, field
from pathlib import Path
from html2md import html_to_markdown
from wxr import (REPO_ROOT, cache_path, clean_media_name, load_items, slugify)
CONTENT = REPO_ROOT / "content"
ASSETS_IMG = REPO_ROOT / "assets" / "img"
STATIC_PDF = REPO_ROOT / "static" / "dokumente"
# WordPress page id -> destination. Anything not listed is skipped.
PAGE_MAP = {
"13": ("aktuelles/_index.md", None),
"72": ("termine.md", "termine"),
"128": ("downloads.md", None),
"138": ("impressum.md", None),
"98": ("archiv/vergangene-o-fahrten.md", None),
"194": ("archiv/40-jahre-mcg/index.md", None),
"217": ("archiv/familienausfahrt-2019/index.md", None),
"287": ("archiv/bilder-familienausfahrt/index.md", None),
"473": ("archiv/familienfahrt-2022/index.md", None),
}
# Deliberately dropped, with the reason recorded for the migration report.
SKIP = {
"3": "stock WordPress privacy-policy draft (English, references an old IP)",
"12": "WP home page body was the placeholder 'Dies ist nur eine Demo Seite'",
"410": "empty draft post",
}
# Site chrome pulled out of the media library into assets/img/.
SITE_IMAGES = {
"Emblem_transparent.png": "logo.png",
"cropped-Banner_small_web.jpg": "banner.jpg",
}
SLUG_FIXES = {
"29-o-fahrt-zum-50-jubilaum": "29-o-fahrt-zum-50-jubilaeum",
"regelmasger-stammtisch": "regelmaessiger-stammtisch",
"vorankundigung-mcg-wochenende-2024": "vorankuendigung-mcg-wochenende-2024",
"o-fahrt-2019": "25-o-fahrt-2022", # title says 25. O-Fahrt 2022; slug was stale
"termin": "28-o-fahrt-2025", # title says 28. O-Fahrt 2025
}
@dataclass
class Target:
item: object
path: Path
bundle: Path | None
layout: str | None = None
images: dict[str, str] = field(default_factory=dict) # attachment id -> filename
# --------------------------------------------------------------------- helpers
def yaml_quote(value: str) -> str:
return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
def iso(date_gmt: str) -> str:
if not date_gmt or date_gmt.startswith("0000"):
return ""
return date_gmt.replace(" ", "T") + "Z"
def referenced_ids(item, by_filename: dict[str, str] | None = None) -> list[str]:
"""Every attachment id an item points at, in document order.
Some posts carry stale ``wp-image-NNN`` ids whose attachment record was
deleted from WordPress even though the upload itself survived (post 81 points
at id 83, which no longer exists). When ``by_filename`` is supplied, such
references are recovered from the ``src`` URL instead.
"""
ids: list[str] = []
def add(value):
if value and value not in ids:
ids.append(value)
thumb = item.meta.get("_thumbnail_id")
add(thumb)
for m in re.finditer(r'
]*>', item.content):
tag = m.group(0)
cls = re.search(r'wp-image-(\d+)', tag)
if cls:
add(cls.group(1))
if by_filename is not None:
src = re.search(r'src="([^"]+)"', tag)
if src:
add(by_filename.get(clean_media_name(src.group(1).rsplit("/", 1)[-1])))
for m in re.finditer(r'', item.content, re.S):
try:
for gid in json.loads(m.group(1)).get("ids", []):
add(str(gid))
except json.JSONDecodeError:
pass
for m in re.finditer(r'', item.content, re.S):
try:
add(str(json.loads(m.group(1)).get("id", "")))
except json.JSONDecodeError:
pass
return [i for i in ids if i]
# ------------------------------------------------------------------ conversion
def convert_body(item, target: Target, attachments: dict, pdf_names: dict) -> str:
body = item.content
# Galleries: replace the whole block with a shortcode listing bundle files.
def gallery_repl(match: re.Match) -> str:
try:
ids = [str(i) for i in json.loads(match.group(1)).get("ids", [])]
except json.JSONDecodeError:
return ""
files = [target.images[i] for i in ids if i in target.images]
if not files:
return ""
args = " ".join(f'"{f}"' for f in files)
return f"\n\n{{{{< gallery {args} >}}}}\n\n"
body = re.sub(
r'.*?',
gallery_repl, body, flags=re.S)
# File blocks: PDFs live in static/dokumente/, linked via the pdf shortcode.
def file_repl(match: re.Match) -> str:
block = match.group(0)
try:
attrs = json.loads(match.group(1))
except json.JSONDecodeError:
return ""
att_id = str(attrs.get("id", ""))
href = attrs.get("href", "")
name = pdf_names.get(att_id) or clean_media_name(href.rsplit("/", 1)[-1])
label = re.search(r'>([^<]+)', block)
label = label.group(1).strip() if label else name
return f'\n\n{{{{< pdf "{name}" "{label}" >}}}}\n\n'
body = re.sub(
r'.*?',
file_repl, body, flags=re.S)
# Shortcodes that have no Hugo equivalent.
body = re.sub(r'.*?', "",
body, flags=re.S)
body = re.sub(r'\[table id=\d+\s*/?\]', "", body)
body = re.sub(r'\[widgets_on_pages[^\]]*\]', "", body)
# Remaining Gutenberg comments carry no information once converted.
body = re.sub(r'', "", body)
# WordPress' editor leaves these around every translated paragraph.
body = re.sub(r'\s*', "", body)
body = re.sub(r'', "", body)
def resolve_image(src: str, cls: str) -> str:
m = re.search(r'wp-image-(\d+)', cls or "")
if m and m.group(1) in target.images:
return target.images[m.group(1)]
# Fall back to matching on the original upload filename.
base = clean_media_name(src.rsplit("/", 1)[-1])
for name in target.images.values():
if name == base:
return name
return src
return html_to_markdown(body, resolve_image=resolve_image)
def front_matter(item, target: Target, extra_tags: list[str]) -> str:
lines = ["---", f"title: {yaml_quote(item.title)}"]
if iso(item.date_gmt):
lines.append(f"date: {iso(item.date_gmt)}")
if iso(item.modified_gmt) and item.modified_gmt != item.date_gmt:
lines.append(f"lastmod: {iso(item.modified_gmt)}")
if extra_tags:
lines.append("tags: [" + ", ".join(yaml_quote(t) for t in extra_tags) + "]")
thumb = item.meta.get("_thumbnail_id")
if thumb and thumb in target.images:
lines.append(f"featured: {yaml_quote(target.images[thumb])}")
if target.layout:
lines.append(f"layout: {yaml_quote(target.layout)}")
lines.append("---")
return "\n".join(lines)
# ------------------------------------------------------------------------ main
def main() -> int:
items = load_items()
attachments = {k: v for k, v in items.items() if v.post_type == "attachment"}
# Cleaned upload filename -> attachment id, for recovering stale wp-image ids.
by_filename: dict[str, str] = {}
for att_id, att in attachments.items():
if att.attachment_url:
by_filename.setdefault(
clean_media_name(att.attachment_url.rsplit("/", 1)[-1]), att_id)
targets: list[Target] = []
for item in items.values():
if item.post_id in SKIP or item.status != "publish":
continue
if item.post_type == "page":
mapping = PAGE_MAP.get(item.post_id)
if not mapping:
print(f"!! unmapped page {item.post_id} ({item.slug})", file=sys.stderr)
continue
rel, layout = mapping
path = CONTENT / rel
elif item.post_type == "post":
slug = SLUG_FIXES.get(item.slug, slugify(item.slug))
path = CONTENT / "aktuelles" / slug / "index.md"
layout = None
else:
continue
bundle = path.parent if path.name == "index.md" else None
targets.append(Target(item=item, path=path, bundle=bundle, layout=layout))
# ---- assign media -------------------------------------------------
used: set[str] = set()
pdf_names: dict[str, str] = {}
for target in targets:
for att_id in referenced_ids(target.item, by_filename):
att = attachments.get(att_id)
if not att or not att.attachment_url:
continue
src = cache_path(att.attachment_url)
if not src.exists():
print(f"!! missing cached media for {att_id}: {att.attachment_url}",
file=sys.stderr)
continue
name = clean_media_name(att.attachment_url.rsplit("/", 1)[-1])
if src.suffix.lower() == ".pdf":
STATIC_PDF.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, STATIC_PDF / name)
pdf_names[att_id] = name
target.images[att_id] = name
else:
if target.bundle is None:
# Non-bundle page (termine.md, impressum.md, ...) -> shared assets.
ASSETS_IMG.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, ASSETS_IMG / name)
target.images[att_id] = f"/img/{name}"
else:
target.bundle.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, target.bundle / name)
target.images[att_id] = name
used.add(att_id)
# Any PDF in the library that no page links to is still worth keeping around
# for the Downloads page; copy them all so nothing is lost.
for att_id, att in attachments.items():
if not att.attachment_url.lower().endswith(".pdf"):
continue
src = cache_path(att.attachment_url)
if src.exists():
STATIC_PDF.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, STATIC_PDF / clean_media_name(
att.attachment_url.rsplit("/", 1)[-1]))
used.add(att_id)
# Site chrome + widget images (members / memoriam) into assets/img/.
ASSETS_IMG.mkdir(parents=True, exist_ok=True)
widget_images = set()
for yaml_file in ("members.yaml", "memoriam.yaml"):
text = (REPO_ROOT / "data" / yaml_file).read_text(encoding="utf-8")
widget_images.update(re.findall(r'image:\s*"img/([^"]+)"', text))
for att_id, att in attachments.items():
original = att.attachment_url.rsplit("/", 1)[-1]
name = clean_media_name(original)
src = cache_path(att.attachment_url)
if not src.exists():
continue
if original in SITE_IMAGES:
shutil.copy2(src, ASSETS_IMG / SITE_IMAGES[original])
used.add(att_id)
elif name in widget_images:
shutil.copy2(src, ASSETS_IMG / name)
used.add(att_id)
# ---- write content ------------------------------------------------
# content/ is the source of truth once imported, so an accidental re-run must
# not throw away hand edits. Pass --force to regenerate anyway.
force = "--force" in sys.argv
written = kept = 0
for target in targets:
item = target.item
target.path.parent.mkdir(parents=True, exist_ok=True)
if target.path.exists() and not force:
kept += 1
continue
tags = ["O-Fahrt"] if any(nice == "ofahrt" for _, nice in item.categories) else []
body = convert_body(item, target, attachments, pdf_names)
target.path.write_text(
front_matter(item, target, tags) + "\n\n" + body + "\n",
encoding="utf-8")
written += 1
print(f"wrote {written} content files"
+ (f", kept {kept} existing (use --force to overwrite)" if kept else ""))
print(f"media used: {len(used)}/{len(attachments)} attachments")
unused = sorted(set(attachments) - used, key=int)
print(f"unreferenced attachments: {len(unused)}")
for att_id in unused:
print(f" - {attachments[att_id].attachment_url}")
return 0
if __name__ == "__main__":
raise SystemExit(main())