fuzzyworld-tools / commit
Blog: static generator for writing.fuzzyworld.net, WordPress importer, tests
PLAN.md | 14 +- README.md | 49 ++ bin/fw | 69 +++ lib/wp_import.py | 1217 +++++++++++++++++++++++++++++++++++++++++++ server/atom.py | 72 +++ server/blog.py | 954 +++++++++++++++++++++++++++++++++ server/site-blog/blog.css | 86 +++ server/site-blog/blog.js | 57 ++ server/site-blog/htaccess | 31 ++ server/site-blog/robots.txt | 18 + tests/test_blog.py | 736 ++++++++++++++++++++++++++ tests/test_wp_import.py | 555 ++++++++++++++++++++ 12 files changed, 3856 insertions(+), 2 deletions(-) diff --git a/PLAN.md b/PLAN.md index 9acd30d..261a7da 100644 --- a/PLAN.md +++ b/PLAN.md @@ -49,7 +49,7 @@ The order goes from cheap to expensive, and each step reuses what the one before **Built 2026-10-28; deployed 2026-10-31** (server, then the TrueNAS runner, then the laptop runner with `--authorize`, in that order: a new runner against the old server has every heartbeat refused). Smoke-tested: both runners poll DreamHost and find nothing to run; no real job has gone through them yet. Review follow-ups (claim check on `done`, - tolerant `release`) are in the working tree, not yet deployed. + tolerant `release`) deployed the same day. 1. **Feeds**: Atom per repo (commits, releases), for app deploys, and site-wide. Smallest step: the data is already gathered for the log pages and the apps index. The blog reuses it. 1b. **Blog** at `blog.fuzzyworld.net` (see [Blog](#blog)): a generator that turns a folder or repo of Markdown into a @@ -98,7 +98,17 @@ always-on machine; a small setup is a DreamHost account and a laptop. `blog.fuzzyworld.net` is WordPress today (theme Scrawl, 144 posts, 4 pages, 45 comments, 15 media files, 3 categories, 16 tags, one author). Goal: the same site, generated by this repo, with no WordPress. -**Rename requested 2026-10-31:** the writing blog moves to `writing.fuzzyworld.net` (queue item Q4 in `~/QUEUE.md`). This section still says `blog.fuzzyworld.net`; it is updated once Pavel confirms the new name and what the old site becomes. +**Built and live 2026-10-31 at `writing.fuzzyworld.net`** (queue item Q4; not linked from anywhere yet). The generator is +`server/blog.py` (+ `server/atom.py`, `server/site-blog/`), the importer `lib/wp_import.py`; commands and behavior are in +the README ("Blog"). Decisions taken: build on the laptop from Dropbox and upload only the audited output folder (never +the source); footer "All rights reserved" notice with an AI-training line and a `robots.txt` blocking AI crawlers; the +WordPress text wins when a source file differs (similarity under 0.95: 100 of the 142 posts are written from WordPress +into `~/Blog/wordpress/`); the old site's custom CSS is carried over (hidden `<br>`, dotted scene breaks, underlined +chapter headings); a light/dark switch and reading-progress bar added (they needed `script-src 'self'`). The Atom +writer exists now, so Next 1 (Feeds) can reuse it. **Still to do:** redirect `blog.fuzzyworld.net` to `writing.` +(Q6, needs a WordPress backup first), link to the site, building from the push hook after the move to a private repo, +the `til` blog (Q14, same importer with `--wp https://til.fuzzyworld.net`). Below, the plan as written before the +build: it still says `blog.fuzzyworld.net` in places; read that as `writing.fuzzyworld.net`. **What is there** (read from its public API, 2026-10-28): - Posts at `/YYYY/MM/DD/<slug>/`, an RSS feed at `/feed/`, and monthly archives. Categories hardly matter ("Uncategorized" diff --git a/README.md b/README.md index 69f776a..1fa6e37 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,8 @@ On the laptop, `bin/fw` runs `deploy` locally and forwards repo commands to Drea | `fw ci <name> [branch]` | queue a CI run of a branch's latest commit (default: the default branch) | | `fw ci-list` | queued and running CI jobs | | `fw ci-watch` | follow CI in the terminal (private repos too): running jobs with each job's state, the queue, the last 5 results; ends when nothing is queued or running | +| `fw blog list\|build\|preview\|deploy <config>` | the static blog (see [Blog](#blog)); runs on the laptop | +| `fw blog import-wp <config> --wp URL` / `fw blog check <config>` | one-time WordPress import / compare the built site with the WordPress text | Push to a repo with `git push dh:git/<name>.git` (`dh` is the SSH alias in `~/.ssh/config`). @@ -101,6 +103,7 @@ Other settings in the repo config: `fuzzyworld.description`, `fuzzyworld.ci` (`o | `~/code.fuzzyworld.net/<name>/` | generated pages for public repos | | `~/code.fuzzyworld.net/<name>.git` | public clone copy (dumb HTTP), refreshed on push | | `~/apps.fuzzyworld.net/<app>/` | deployed apps (`fw deploy`); `index.html` at the root is generated | +| `~/writing.fuzzyworld.net/` | the blog's built site, uploaded by `fw blog deploy` from the laptop (never edited there) | | `~/fw/tools/server/` | this repo's `server/` directory, installed by `server/install.sh` | | `~/bin/fw` | symlink to `~/fw/tools/server/fw` | | `~/fw/state/` | per-repo page manifests and locks | @@ -267,6 +270,52 @@ pictures go to `apps.fuzzyworld.net/_shots/<app>/` and appear on the apps index Without runner screenshots, the images in a public repo's `screenshots/` folder are shown instead. +## Blog + +A generic static-blog generator: a folder of Markdown files and one JSON config become a site (`server/blog.py`). +It runs on the **laptop**, builds into a clean temporary folder and uploads only that folder with `rsync`, so +nothing in the source folder (drafts, notes, config) is ever published. It is used for `writing.fuzzyworld.net`. + +``` +fw blog list ~/Blog/fw-blog.json # every source file: URL, title, date, tags, published or not +fw blog build ~/Blog/fw-blog.json --out DIR [--no-htaccess] # write the site into DIR (other options are refused) +fw blog preview ~/Blog/fw-blog.json [port] # build and serve it locally (no redirects, no feed content type) +fw blog deploy ~/Blog/fw-blog.json --dry-run # then, for real: --yes (or answer the prompt) +``` + +- **What is published:** only posts with `published: true` (front matter in the file, or the config's `posts` table keyed + by file name; front matter wins over the table). Pages (files in `pages_dir`) are opt-in the same way: a page needs + `published: true` in its front matter (the importer writes it). Text under `## Spelling` and + `## Removed writing` and every HTML comment (except `<!--more-->`) never reach the output. The build stops (rather + than risk a leak) when a heading that looks private is written another way (`##Spelling`, an underlined heading) or a + code fence is never closed. +- **Safety checks:** slugs (posts, pages, tags, the feed path) may use only letters, digits and `. _ % -`; values that + go into `.htaccess` (`redirects`, `wp_ids`, `home`, `feed.path`) may not contain spaces or control characters, even + percent-encoded; `deploy.dir` must be a site folder such as `writing.fuzzyworld.net` (relative to the login folder; + `~`, `.`, `/`, `..` are refused); symbolic links in the media folder are skipped. +- **Config:** `title`, `base_url` and `source` are required; unknown keys are an error. Others: `tagline`, `author`, + `copyright` (`{years}` and `{author}` are filled in), `body_section`, `line_breaks`, `url`, `home`, `pages_dir`, `nav`, `tags`, `feed`, + `reading_wpm`, `media`, `rewrite_hosts`, `redirects`, `wp_ids`, `comments`, `timezone`, `deploy`, `posts`; see the + top of `server/blog.py`. `source` and `comments` are relative to the config file's folder; `pages_dir` and `media` + are relative to `source`. +- **Output:** posts at `/YYYY/MM/DD/<slug>/`, `/blog/` (every post by year and month, with excerpts and a "Continue + reading" link), `/tag/<slug>/`, pages, an Atom feed at `/feed/` (`server/atom.py`), `.htaccess` (redirects from + `redirects` and `wp_ids`, no script execution, a security policy that allows only the site's own `blog.js`), + `robots.txt` that blocks AI crawlers, comments as static text. +- **Reader features** (`server/site-blog/blog.js`): a light/dark switch (the choice is kept in the browser; without one the + system setting decides) and a reading-progress bar on posts. +- **Upload safety:** `deploy` audits the built folder first and refuses anything that is not a generated file or an + allowlisted media type (`png jpg jpeg gif webp svg pdf epub kpf mp3 mp4 zip`), then rsyncs only that folder (it + keeps `.well-known`). It asks for confirmation on every real upload; `--dry-run` shows what would be sent. +- **WordPress import** (`lib/wp_import.py`, `fw blog import-wp`): reads the public REST API, skips password-protected + posts, matches each post to a source file (similarity of 0.95 or more uses your file; otherwise the WordPress text is + written to `wordpress/<date>-<slug>.md`; a non-ASCII slug gives a readable file name but keeps its URL), and writes + comments (name, date and text only, into the `comments` file next to the config), pages (with `published: true`), + media and the config entries. `--dry-run` writes only a cache and `wordpress/_report.txt`; a second run changes nothing. + `fw blog check` lists posts whose built text differs from WordPress. +- Python-Markdown and Pygments are installed into `~/fw/pylib` on the laptop the first time `fw blog` runs. +- Tests: `python3 -m unittest discover tests`. + ## VS Code in the browser `apps.fuzzyworld.net/vscode/?repo=<name>` (and `&branch=<branch>` for an unmerged branch) opens a public repo in diff --git a/bin/fw b/bin/fw index 1cd79dd..96c11b5 100755 --- a/bin/fw +++ b/bin/fw @@ -4,6 +4,11 @@ # fw repoint [DIR] [--https] [--dry-run] move working copies' origin from GitLab to DreamHost # fw list | new <name> [--public] [--description TEXT] | public <name> | private <name> # fw usage [--all|--now] Claude Code token use per session (see lib/usage.py) +# fw blog list|build|preview|deploy <config> static blog from Markdown (see server/blog.py); +# build takes --out DIR and --no-htaccess +# fw blog import-wp <config> --wp URL [--dry-run] [--force] [--refresh] [--probe-links] +# import the WordPress blog (see lib/wp_import.py) +# fw blog check <config> [--out DIR] compare the built site with the cached WordPress text # fw describe <name> TEXT | publish <name> | publish-all | ci <name> [branch] | ci-list | ci-watch # (run on DreamHost over ssh) set -euo pipefail @@ -11,7 +16,71 @@ HOST=${FW_HOST:-dh} # Follow the symlink npm creates in node_modules/.bin to find this package. HERE=$(cd "$(dirname "$(readlink -f "${BASH_SOURCE[0]}")")/.." && pwd) +# fw blog: runs on the laptop. Markdown and Pygments live in ~/fw/pylib (installed here when missing). +blog() { + local sub=${1:-} config=${2:-} tmp=${TMPDIR:-/tmp} + tmp=${tmp%/} + local pylib=${FW_PYLIB:-$HOME/fw/pylib} blogpy="$HERE/server/blog.py" + case "$sub" in + list|build|preview|deploy|import-wp|check) ;; + *) echo "usage: fw blog list|build|preview|deploy|import-wp|check <config> [options]" >&2; exit 2 ;; + esac + [ -n "$config" ] || { echo "fw blog $sub: needs the config file (e.g. ~/Blog/fw-blog.json)" >&2; exit 2; } + shift 2 + if ! PYTHONPATH="$pylib" python3 -c 'import markdown, pygments' 2>/dev/null; then + echo "fw: installing Markdown and Pygments into $pylib" >&2 + python3 -m pip install --quiet --target "$pylib" Markdown==3.7 Pygments==2.21.0 + fi + case "$sub" in + import-wp) + # Reads the WordPress REST API (cached in <source>/wordpress/_api/); --dry-run writes only that cache and the report. + exec python3 "$HERE/lib/wp_import.py" import "$config" "$@" ;; + check) + exec python3 "$HERE/lib/wp_import.py" check "$config" "$@" ;; + list) + python3 "$blogpy" list "$config" ;; + build) + local out=$tmp/fw-blog-build extra=() + while [ $# -gt 0 ]; do + case "$1" in + --out) out=${2:?--out needs a folder}; shift 2 ;; + --no-htaccess) extra+=(--no-htaccess); shift ;; + *) echo "fw blog build: unknown option $1 (use --out DIR, --no-htaccess)" >&2; exit 2 ;; + esac + done + python3 "$blogpy" build "$config" --out "$out" ${extra[@]+"${extra[@]}"} ;; + preview) + # No .htaccess here (python's server ignores it, so redirects are not tested); the feed is at /feed/index.xml. + local out=$tmp/fw-blog-preview port=${1:-8000} + python3 "$blogpy" build "$config" --out "$out" --no-htaccess + echo "preview: http://127.0.0.1:$port/ (ctrl-c to stop)" + exec python3 -m http.server --bind 127.0.0.1 --directory "$out" "$port" ;; + deploy) + # Uploads ONLY a freshly built, audited output folder: never the source folder. + local dry=0 yes=0 a + for a in "$@"; do + case "$a" in --dry-run) dry=1 ;; --yes) yes=1 ;; *) echo "fw blog deploy: unknown option $a" >&2; exit 2 ;; esac + done + local out=$tmp/fw-blog-deploy target + target=$(python3 "$blogpy" deploy-target "$config") + python3 "$blogpy" build "$config" --out "$out" + python3 "$blogpy" audit "$out" + local rsync_opts=(-rlptz --delete --exclude .well-known --exclude .dh-diag --exclude .fw-blog-out) + if [ "$dry" = 1 ]; then + exec rsync -n -v "${rsync_opts[@]}" "$out/" "$target/" + fi + if [ "$yes" != 1 ]; then + [ -t 0 ] || { echo "fw blog deploy: not a terminal; pass --yes to upload to $target" >&2; exit 1; } + read -r -p "Upload the output above to $target, deleting other files there? [y/N] " reply + [ "$reply" = y ] || [ "$reply" = Y ] || { echo "not uploaded" >&2; exit 1; } + fi + exec rsync "${rsync_opts[@]}" "$out/" "$target/" ;; + esac +} + case "${1:-help}" in + blog) + shift; blog "$@" ;; deploy) shift; exec node "$HERE/lib/deploy.mjs" "$@" ;; usage) diff --git a/lib/wp_import.py b/lib/wp_import.py new file mode 100644 index 0000000..cf6ef0f --- /dev/null +++ b/lib/wp_import.py @@ -0,0 +1,1217 @@ +#!/usr/bin/env python3 +"""WordPress importer and checker for the static blog (laptop only, standard library only). + + wp_import.py import <config> --wp URL [--dry-run] [--force] [--refresh] [--probe-links] + [--cache DIR] [--no-media] [--media-from SRC] [--timezone TZ] [--home SLUG] + wp_import.py check <config> [--out DIR] [--cache DIR] + +`fw blog import-wp` and `fw blog check` are the usual way in. The importer reads the WordPress REST API +(posts, pages, comments, tags, media) and caches the raw JSON in <source>/wordpress/_api/ (a later run uses +the cache; --refresh fetches again; --cache DIR reads or fills another folder). It matches each public post +to a Markdown file in <source> by comparing normalized word sequences (difflib ratio >= 0.95; each file is +used once). A post with no match is written to <source>/wordpress/<YYYY-MM-DD>-<slug>.md as Markdown made +from its HTML. Pages go to <source>/<pages_dir>/ (each with `published: true` in its front matter: pages, like posts, are opt-in), comments to +the `comments` file (relative to the config file's folder, like the generator reads it; name, date and text only), +images and files are copied from the old server with a read-only rsync, and the config receives `posts`, +`tags`, `redirects`, `wp_ids` and a few other entries without overwriting existing ones (unless --force). + +--dry-run writes only <source>/wordpress/_api/ and <source>/wordpress/_report.txt. Everything else is written +inside <source> and the config file only. Password-protected posts and their comments are skipped and listed +in the report. --probe-links sends HEAD requests (to the old site) for internal links that the new site will +not generate, and records the 301 targets as `redirects`. + +`check` builds the site into a temporary folder (or reads --out), compares each built post with the cached +WordPress text, compares <br> counts, and verifies that every old post, page and tag URL and every redirect +target exists in the output. It exits with status 1 when it finds a problem. +""" + +import argparse +import collections +import datetime +import difflib +import glob +import html +import html.parser +import json +import os +import re +import shutil +import subprocess +import sys +import tempfile +import unicodedata +import urllib.error +import urllib.parse +import urllib.request +import zoneinfo + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.join(HERE, "..", "server")) +import blog # noqa: E402 + +MATCH_THRESHOLD = 0.95 # a source file with at least this word-sequence similarity is the post's source +CHECK_THRESHOLD = 0.98 # `check` lists built posts below this similarity to the WordPress text +USER_AGENT = "fw-blog-import/1 (static blog migration)" +API_NAMES = ("posts", "pages", "comments", "tags", "media") +# Names the development cache used before the importer existed (one file per page of results). +LEGACY_CACHE = {"posts": ("p[0-9]*.json",), "comments": ("c.json",)} + + +class WpImportError(blog.BlogError): + """A problem reported without a traceback.""" + + +# ---------------------------------------------------------------- HTML -> tree + +VOID = {"br", "hr", "img", "meta", "link", "input", "area", "base", "col", "embed", "source", "track", "wbr"} +BLOCK = {"p", "div", "h1", "h2", "h3", "h4", "h5", "h6", "blockquote", "ul", "ol", "li", "pre", "hr", "figure", + "figcaption", "table", "thead", "tbody", "tr", "td", "th", "section", "article", "header", "footer", + "aside", "center", "address", "dl", "dt", "dd", "form", "main", "nav", "details", "summary"} +KEEP_TAGS = {"del", "ins", "sup", "sub", "u", "s", "small", "mark", "kbd"} # inline tags kept as HTML in Markdown +DROP_CLASSES = {"sharedaddy", "jp-relatedposts", "jetpack-likes-widget-wrapper"} + + +class Node: + __slots__ = ("tag", "attrs", "children", "text") + + def __init__(self, tag, attrs=None, text=None): + self.tag, self.attrs, self.children, self.text = tag, dict(attrs or {}), [], text + + +class _Builder(html.parser.HTMLParser): + def __init__(self): + super().__init__(convert_charrefs=True) + self.root = Node("#root") + self.stack = [self.root] + + def handle_starttag(self, tag, attrs): + node = Node(tag, {k: (v if v is not None else "") for k, v in attrs}) + self.stack[-1].children.append(node) + if tag not in VOID: + self.stack.append(node) + + def handle_startendtag(self, tag, attrs): + self.handle_starttag(tag, attrs) + if tag not in VOID: + self.handle_endtag(tag) + + def handle_endtag(self, tag): + for i in range(len(self.stack) - 1, 0, -1): + if self.stack[i].tag == tag: + del self.stack[i:] + return + + def handle_data(self, data): + self.stack[-1].children.append(Node(None, text=data)) + + def handle_comment(self, data): + self.stack[-1].children.append(Node("#comment", text=data)) + + +def parse_html(text): + b = _Builder() + b.feed(text or "") + b.close() + return b.root + + +def classes(node): + return (node.attrs.get("class") or "").split() + + +def is_dropped(node): + """What WordPress adds around a post and the page does not need: scripts, the reading-time plugin's span, + share buttons, and every HTML comment except <!--more--> (comments may hold text that was cut on purpose).""" + tag = node.tag + if tag == "#comment": + return node.text.strip() != "more" + if tag in ("script", "style", "noscript", "head", "title"): + return True + cl = classes(node) + if tag == "span" and ("rt-reading-time" in cl or "span-reading-time" in cl): + return True + if node.attrs.get("id") == "bsf_rt_marker": + return True + if tag == "span" and str(node.attrs.get("id", "")).startswith("more-"): + return True + return bool(DROP_CLASSES & set(cl)) + + +def youtube_url(src): + m = re.match(r"(?:https?:)?//(?:www\.)?youtube(?:-nocookie)?\.com/embed/([\w-]+)", src or "") + return f"https://www.youtube.com/watch?v={m.group(1)}" if m else None + + +def embed_of(node): + """(link text, URL) for an embed the site cannot show (its CSP blocks frames and scripts): a YouTube frame + and a Flickr embed become a plain link. None for anything else.""" + if node.tag == "iframe": + url = youtube_url(node.attrs.get("src")) + return (f"Video: {url}", url) if url else None + if node.tag == "a" and "data-flickr-embed" in node.attrs and node.attrs.get("href"): + alt = next((c.attrs.get("alt") for c in node.children if c.tag == "img" and c.attrs.get("alt")), None) + return (node.attrs.get("title") or alt or "Photo on Flickr", node.attrs["href"]) + return None + + +# ---------------------------------------------------------------- tree -> Markdown + +def collapse(text): + return re.sub(r"[ \t\r\n\f ]+", " ", text) + + +def esc_text(text): + text = text.replace("\\", "\\\\") + text = re.sub(r"([*`\[\]])", r"\\\1", text) + text = re.sub(r"(?<![A-Za-z0-9])_|_(?![A-Za-z0-9])", r"\\_", text) + text = re.sub(r"&(?=#?\w+;)", "&", text) + return text.replace("<", "<") + + +def esc_line_starts(text): + out = [] + for line in text.split("\n"): + m = re.match(r"(\d+)([.)])(\s|$)", line) + if m: + line = f"{m.group(1)}\\{m.group(2)}" + line[m.end(2):] + elif re.match(r"(#|>|[-+](\s|$)|-{2,}\s*$)", line): + line = "\\" + line + out.append(line) + return "\n".join(out) + + +def finish_inline(text, single_line=False): + """Tidy one paragraph's worth of inline Markdown. A "\\n" in it came from a <br>: the first of several in a + row stays a newline, the others become literal <br> so the count of <br> is kept.""" + text = text.strip(" \n") + text = re.sub(r"(?: *\n)+ *", lambda m: "\n" + "<br>" * (m.group(0).count("\n") - 1), text) + text = re.sub(r" {2,}", " ", text) + if single_line: + text = re.sub(r"\s*\n\s*", " ", text) + return esc_line_starts(text) + + +def wrap_marks(marker, content): + """*content*, one marker pair per line (so an emphasis never spans a <br>), spaces kept outside.""" + out = [] + for line in content.split("\n"): + core = line.strip() + out.append(line[:len(line) - len(line.lstrip())] + marker + core + marker + line[len(line.rstrip()):] if core else line) + return "\n".join(out) if content.strip() else "" + + +def link_target(url): + return url.strip().replace(" ", "%20").replace("(", "%28").replace(")", "%29") + + +class Converter: + def __init__(self, notes=None): + self.notes = notes if notes is not None else [] + + def inline(self, node): + if node.tag is None: + return esc_text(collapse(node.text)) + if is_dropped(node): + return "" + tag = node.tag + if tag == "#comment": + return "" + if tag == "br": + return "\n" + if tag == "img": + src = node.attrs.get("src") + alt = re.sub(r"[\[\]]", "", collapse(node.attrs.get("alt", "")).strip()) + return f"})" if src else "" + embed = embed_of(node) + if embed: + return f"[{esc_text(embed[0])}]({link_target(embed[1])})" + if tag == "iframe": + self.notes.append(f"iframe dropped: {node.attrs.get('src', '?')}") + return "" + inner = "".join(self.inline(c) for c in node.children) + if tag == "a": + href = node.attrs.get("href", "").strip() + if not href or href.lower().startswith("javascript:"): + return inner + return f"[{inner}]({link_target(href)})" if inner.strip() else "" + if tag in ("em", "i"): + return wrap_marks("*", inner) + if tag in ("strong", "b"): + return wrap_marks("**", inner) + if tag == "code": + text = "".join(self.raw_text(c) for c in node.children) + text = collapse(text).strip() + if not text: + return "" + fence = "`" * (max([len(r) for r in re.findall(r"`+", text)] + [0]) + 1) + pad = " " if text.startswith("`") or text.endswith("`") else "" + return f"{fence}{pad}{text}{pad}{fence}" + if tag in KEEP_TAGS: + return f"<{tag}>{inner}</{tag}>" if inner.strip() else "" + if tag in BLOCK: + return f" {inner} " + return inner + + def raw_text(self, node): + if node.tag is None: + return node.text + if is_dropped(node): + return "" + if node.tag == "br": + return "\n" + return "".join(self.raw_text(c) for c in node.children) + + def blocks(self, children): + """The Markdown blocks (strings) for a list of nodes.""" + out, run = [], [] + + def flush(): + text = finish_inline("".join(run)) + run.clear() + if text: + out.append(text) + + for node in children: + if node.tag is not None and is_dropped(node): + continue + if node.tag == "#comment": # only <!--more--> survives is_dropped + flush() + out.append("<!--more-->") + elif node.tag is None or node.tag not in BLOCK and node.tag != "iframe": + run.append(self.inline(node)) + elif node.tag == "iframe": + flush() + text = self.inline(node) + if text: + out.append(text) + else: + flush() + out.extend(self.block(node)) + flush() + return out + + def block(self, node): + tag = node.tag + if tag == "hr": + return ["---"] + if re.fullmatch(r"h[1-6]", tag): + # a `## ` line would end the post body, so no heading is higher than ### + text = finish_inline("".join(self.inline(c) for c in node.children), single_line=True) + return [f"{'#' * max(3, int(tag[1]))} {text}"] if text else [] + if tag == "pre": + return self.pre(node) + if tag in ("ul", "ol"): + text = self.list(node, tag == "ol") + return [text] if text else [] + if tag == "blockquote": + inner = "\n\n".join(self.blocks(node.children)) + return ["\n".join(f"> {line}" if line else ">" for line in inner.split("\n"))] if inner else [] + return self.blocks(node.children) + + def pre(self, node): + text = self.raw_text(node).strip("\n") + if not text.strip(): + return [] + code = next((c for c in node.children if c.tag == "code"), None) + lang = next((c[9:] for c in classes(code) if c.startswith("language-")), "") if code else "" + fence = "`" * max(3, max([len(r) for r in re.findall(r"`+", text)] + [0]) + 1) + return [f"{fence}{lang}\n{text}\n{fence}"] + + def list(self, node, ordered): + items, loose = [], False + for n, li in enumerate((c for c in node.children if c.tag == "li" and not is_dropped(c)), 1): + blocks = self.blocks(li.children) + if not blocks: + continue + text = blocks[0] + for b in blocks[1:]: + is_list = re.match(r"(- |\d+\. )", b) is not None + loose = loose or not is_list + text += ("\n" if is_list else "\n\n") + b + text = text.replace("\n", "\n ") + items.append((f"{n}. " if ordered else "- ") + re.sub(r"\n {4}(?=\n)", "\n", text)) + return ("\n\n" if loose else "\n").join(items) + + +def html_to_markdown(text, notes=None): + """Markdown for WordPress HTML: <br> becomes a line break, </p> a blank line; see Converter for the rest.""" + conv = Converter(notes) + return "\n\n".join(conv.blocks(parse_html(text).children)).strip() + + +def html_text(text): + """The words a reader sees, as one string (embed links included, scripts and comments not).""" + out = [] + + def walk(node): + if node.tag is None: + out.append(node.text) + return + if is_dropped(node) or node.tag in ("img", "iframe") and not embed_of(node): + return + embed = embed_of(node) + if embed: + out.append(f" {embed[0]} ") + return + sep = " " if node.tag in BLOCK or node.tag == "br" else "" + out.append(sep) + for c in node.children: + walk(c) + out.append(sep) + + walk(parse_html(text)) + return "".join(out) + + +def html_br_count(text): + def count(node): + if node.tag is not None and is_dropped(node): + return 0 + return (1 if node.tag == "br" else 0) + sum(count(c) for c in node.children) + return count(parse_html(text)) + + +def html_plain(text): + """Plain text with paragraphs and line breaks kept (for comments: no markup survives).""" + out = [] + + def walk(node): + if node.tag is None: + out.append(collapse(node.text)) + elif node.tag == "br": + out.append("\n") + elif not is_dropped(node): + sep = "\n\n" if node.tag in BLOCK else "" + out.append(sep) + for c in node.children: + walk(c) + out.append(sep) + + walk(parse_html(text)) + s = re.sub(r" *\n *", "\n", "".join(out)) + return re.sub(r"\n{3,}", "\n\n", s).strip() + + +# ---------------------------------------------------------------- comparing texts + +def norm_words(text): + """Words only: lower case, no punctuation, no apostrophes (WordPress turns ' into ’ and -- into –).""" + text = unicodedata.normalize("NFKC", text).lower() + text = re.sub(r"['’‘ʼ`]", "", text) + return re.findall(r"[^\W_]+", text) + + +def md_words(markdown_text): + """The words of a Markdown source (links keep their text, images and URLs are dropped).""" + text = re.sub(r"!\[[^\]]*\]\([^)]*\)", " ", markdown_text) + text = re.sub(r"\]\([^)]*\)", "] ", text) + text = re.sub(r"<https?://[^>]*>", " ", text) + text = re.sub(r"<[^>]+>", " ", text) + return norm_words(html.unescape(text)) + + +def similarity(a, b): + if not a and not b: + return 1.0 + return difflib.SequenceMatcher(None, a, b, autojunk=False).ratio() + + +def title_text(rendered): + return collapse(html.unescape(re.sub(r"<[^>]+>", "", rendered or ""))).strip() + + +# ---------------------------------------------------------------- fetching + +class _NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, *args, **kwargs): + return None + + +class Fetcher: + """The REST API and its cache. `cache` is a folder of <name>.json files; a file there is used instead of + fetching unless `refresh`. `get_json(url) -> (data, total_pages)` and `head(url) -> (status, location)` can be + replaced (tests do).""" + + def __init__(self, wp, cache, refresh=False, get_json=None, head=None, log=lambda s: None): + self.wp, self.cache, self.refresh, self.log = wp.rstrip("/"), cache, refresh, log + self.get_json = get_json or self._get_json + self.head = head or self._head + + def _get_json(self, url): + req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"}) + with urllib.request.urlopen(req, timeout=60) as r: + return json.load(r), int(r.headers.get("X-WP-TotalPages") or 0) + + def _head(self, url): + opener = urllib.request.build_opener(_NoRedirect) + req = urllib.request.Request(url, method="HEAD", headers={"User-Agent": USER_AGENT}) + try: + with opener.open(req, timeout=30) as r: + return r.status, None + except urllib.error.HTTPError as e: + return e.code, e.headers.get("Location") + except (urllib.error.URLError, OSError) as e: + return 0, str(e) + + def path(self, name): + return os.path.join(self.cache, name + ".json") + + def cached(self, name): + """The cached list for `name`, or None.""" + if os.path.exists(self.path(name)): + with open(self.path(name), encoding="utf-8") as f: + return json.load(f) + files = sorted(f for pat in LEGACY_CACHE.get(name, ()) for f in glob.glob(os.path.join(self.cache, pat))) + if files: + data = [] + for f in files: + with open(f, encoding="utf-8") as fh: + data += json.load(fh) + return data + return None + + def collection(self, name): + if not self.refresh: + data = self.cached(name) + if data is not None: + return data + data, page = [], 1 + while True: + order = "id" if name == "tags" else "date" # the tags endpoint has no date to order by (HTTP 400) + url = f"{self.wp}/wp-json/wp/v2/{name}?per_page=100&page={page}&orderby={order}&order=asc" + try: + chunk, total = self.get_json(url) + except urllib.error.HTTPError as e: + if e.code == 400 and page > 1: # past the last page + break + raise WpImportError(f"{url}: HTTP {e.code}") + except (urllib.error.URLError, OSError, ValueError) as e: + raise WpImportError(f"{url}: {e}") + data += chunk + if len(chunk) < 100 or (total and page >= total): + break + page += 1 + os.makedirs(self.cache, exist_ok=True) + atomic_write(self.path(name), json.dumps(data, ensure_ascii=False, indent=1) + "\n") + self.log(f"fetched {len(data)} {name}") + return data + + +def atomic_write(path, text): + os.makedirs(os.path.dirname(path), exist_ok=True) + tmp = path + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + f.write(text) + os.replace(tmp, path) + + +# ---------------------------------------------------------------- reading the API data + +def rendered(field): + return field.get("rendered", "") if isinstance(field, dict) else (field or "") + + +def is_protected(item): + return bool((item.get("content") or {}).get("protected")) or "post-password-required" in (item.get("class_list") or []) + + +def safe_slug(slug, wp_id): + """A file name for a WordPress slug. WordPress sends non-ASCII slugs percent-encoded (caf%c3%a9): the file gets + the readable name (café). The slug itself stays as WordPress has it, because that is the URL.""" + slug = unicodedata.normalize("NFC", urllib.parse.unquote(str(slug or ""))) + slug = re.sub(r"[^\w.-]", "", slug).lstrip(".") + return slug or f"post-{wp_id}" + + +def link_path(link, wp): + """/path/ of a WordPress link (the host is dropped; a query is kept).""" + u = urllib.parse.urlsplit(link or "") + return (u.path or "/") + (f"?{u.query}" if u.query else "") + + +def read_posts(fetcher, notes): + """(public posts, skipped protected posts, other) from the API cache; each public post is a dict.""" + tags = {t["id"]: t for t in fetcher.collection("tags")} + posts, skipped, ignored = [], [], [] + for item in fetcher.collection("posts"): + title = title_text(rendered(item.get("title"))) + if item.get("status", "publish") != "publish": + ignored.append(f'{title} ({item.get("status")})') + elif is_protected(item): + skipped.append({"id": item["id"], "title": title}) + else: + body = rendered(item.get("content")) + posts.append({ + "id": item["id"], "title": title, "slug": str(item["slug"]), + "date": datetime.datetime.fromisoformat(item["date"]), + "date_gmt": item.get("date_gmt"), "html": body, + "path": link_path(item.get("link"), fetcher.wp), + "tags": [tags[t]["slug"] for t in item.get("tags", []) if t in tags], + "tag_names": {tags[t]["slug"]: html.unescape(tags[t]["name"]) for t in item.get("tags", []) if t in tags}, + "text": html_text(body)}) + posts[-1]["words"] = norm_words(posts[-1]["text"]) + return posts, skipped, ignored + + +def read_pages(fetcher): + pages, skipped = [], [] + for item in fetcher.collection("pages"): + title = title_text(rendered(item.get("title"))) + if item.get("status", "publish") != "publish": + continue + if is_protected(item): + skipped.append(title) + continue + pages.append({"id": item["id"], "title": title, "slug": str(item["slug"]), "html": rendered(item.get("content")), + "path": link_path(item.get("link"), fetcher.wp)}) + return pages, skipped + + +# ---------------------------------------------------------------- matching posts to source files + +def source_candidates(cfg): + """Every post file except what the importer wrote itself, with its words.""" + out = [] + for p in blog.scan_posts(cfg): + if p.rel.startswith("wordpress/"): + continue + body = blog.extract_body(p.text, cfg["body_section"], p.body_until, p.rel) + out.append({"rel": p.rel, "post": p, "words": md_words(body), "title_words": norm_words(p.title), + "date": p.date or datetime.datetime.min}) + return out + + +def match_posts(posts, files, threshold=MATCH_THRESHOLD): + """Assign source files to posts. Returns (assigned {post index: (score, file index)}, best {post index: + [(score, file index)] by score}, conflicts [text]). + + Candidates are ranked by score, then equal titles, then files dated on or before the publish date and nearer + to it. Each file is used once; a file or post with several candidates at or above the threshold is reported.""" + best, above = {}, [] + for i, post in enumerate(posts): + sm = difflib.SequenceMatcher(None, autojunk=False) + sm.set_seq2(post["words"]) + ranked = [] + for j, f in enumerate(files): + if not f["words"] or not post["words"]: + continue + sm.set_seq1(f["words"]) + if sm.real_quick_ratio() >= 0.5: + q = sm.quick_ratio() + if q >= 0.5: + ranked.append((q, j)) + ranked.sort(reverse=True) + scored = [] + for n, (q, j) in enumerate(ranked): + if q < threshold and n >= 3: + break + sm.set_seq1(files[j]["words"]) + scored.append((sm.ratio(), j)) + scored.sort(reverse=True) + best[i] = scored + above += [(s, i, j) for s, j in scored if s >= threshold] + + def key(t): + score, i, j = t + f, p = files[j], posts[i] + late = f["date"].date() > p["date"].date() + return (-round(score, 4), f["title_words"] != norm_words(p["title"]), late, + abs((f["date"].date() - p["date"].date()).days), f["rel"], p["path"]) + + assigned, used = {}, set() + for score, i, j in sorted(above, key=key): + if i not in assigned and j not in used: + assigned[i] = (score, j) + used.add(j) + conflicts = [] + claims = collections.defaultdict(list) + for score, i, j in above: + claims[j].append((score, i)) + for j in sorted(claims, key=lambda j: files[j]["rel"]): + if len(claims[j]) > 1: + who = ", ".join(f'"{posts[i]["title"]}" ({s:.3f})' for s, i in sorted(claims[j], reverse=True)) + winner = next((posts[i]["title"] for i, (s, jj) in assigned.items() if jj == j), None) + conflicts.append(f'{files[j]["rel"]} is claimed by {who}; given to "{winner}"') + for i in sorted(best): + cands = [(s, j) for s, j in best[i] if s >= threshold] + if len(cands) > 1: + conflicts.append(f'"{posts[i]["title"]}" matches several files: ' + + ", ".join(f'{files[j]["rel"]} ({s:.3f})' for s, j in cands)) + return assigned, best, conflicts + + +# ---------------------------------------------------------------- planned changes + +class Plan: + """Everything the import would write: files, comments, config entries.""" + + def __init__(self): + self.files = {} # path relative to source -> text + self.post_entries = {} # relative path -> config `posts` entry + self.comments = {} # post path -> [comment] + self.tags = {} # slug -> name + self.wp_ids = {} # str id -> path + self.redirects = {} # old path -> new path + self.lines = collections.defaultdict(list) # report sections + + +def date_text(when): + return when.strftime("%Y-%m-%dT%H:%M:%S") + + +def post_markdown(post, notes): + body = html_to_markdown(post["html"], notes) + return f"# {post['title']}\n\n## Writing\n\n{body}\n" + + +def page_markdown(page, notes): + """A page is published only with `published: true`; its slug is set when the file name cannot say it.""" + front = ["published: true"] + if safe_slug(page["slug"], page["id"]) != page["slug"]: + front.append(f"slug: {page['slug']}") + return f"---\n{chr(10).join(front)}\n---\n# {page['title']}\n\n{html_to_markdown(page['html'], notes)}\n" + + +def build_plan(cfg, posts, pages, comments, assigned, files, notes, home): + plan = Plan() + src = cfg["_source"] + existing_titles = {f["rel"]: f["post"].title for f in files} + for i, post in enumerate(posts): + entry = {"published": True, "date": date_text(post["date"]), "slug": post["slug"], "wp_id": post["id"]} + if post["tags"]: + entry["tags"] = list(post["tags"]) + if i in assigned: + _, j = assigned[i] + rel = files[j]["rel"] + if norm_words(existing_titles[rel]) != norm_words(post["title"]): + entry["title"] = post["title"] + else: + slug = safe_slug(post["slug"], post["id"]) + entry["slug"] = post["slug"] + rel = f"wordpress/{post['date'].strftime('%Y-%m-%d')}-{slug}.md" + plan.files[rel] = post_markdown(post, notes) + plan.post_entries[rel] = entry + post["rel"] = rel + plan.wp_ids[str(post["id"])] = post["path"] + for slug, name in post["tag_names"].items(): + plan.tags[slug] = name + post_paths = {str(p["id"]): p["path"] for p in posts} + for page in pages: + if page["slug"] == "blog": + plan.lines["notes"].append('page "blog" not imported: /blog/ is generated') + continue + if page["slug"] == home: + page["path"] = "/" + plan.files[f"{blog.pages_dir(cfg) or 'pages'}/{safe_slug(page['slug'], page['id'])}.md"] = page_markdown(page, notes) + plan.wp_ids[str(page["id"])] = page["path"] + for c in comments: + if c.get("status", "approved") != "approved" or c.get("type", "comment") != "comment": + continue + path = post_paths.get(str(c.get("post"))) + if path is None: + continue # a protected or unknown post (counted by the caller) + plan.comments.setdefault(path, []).append({ + "id": int(c["id"]), "parent": int(c.get("parent") or 0), + "author": html.unescape(c.get("author_name") or ""), + "date": c.get("date", ""), "text": html_plain(rendered(c.get("content")))}) + for lst in plan.comments.values(): + lst.sort(key=lambda c: (c["date"], c["id"])) + return plan + + +# ---------------------------------------------------------------- links and redirects + +LINK_RE = re.compile(r"""(?:href|src)\s*=\s*["']([^"']+)["']|\]\(\s*<?([^)\s>]+)|(https?://[^\s<>"')\]]+)""") + + +def internal_links(text, hosts): + """Internal links (as /path or /path?query, without #fragment) found in HTML or Markdown.""" + found = set() + for m in LINK_RE.finditer(text or ""): + url = html.unescape(next(g for g in m.groups() if g)).strip().rstrip(".,;:") + for h in hosts: + if url.startswith(h) and url[len(h):len(h) + 1] in ("", "/", "?", "#"): + url = url[len(h):] or "/" + break + if not url.startswith("/") or url.startswith("//"): + continue + found.add(re.sub(r"#.*$", "", url) or "/") + return found + + +def norm_path(path): + p = urllib.parse.unquote(path.split("?")[0].split("#")[0]) + return p if p.endswith("/") or "." in p.rsplit("/", 1)[-1] else p + "/" + + +def generated_paths(cfg, posts, pages, plan, home): + paths = {"/", "/blog/", "/feed/", "/404.html"} | {p["path"] for p in posts} + paths |= {pg["path"] for pg in pages if pg["slug"] != "blog"} + paths |= {f"/tag/{blog.tag_slug(cfg, t)}/" for t in plan.tags} + if home: + paths.add(f"/{home}/") # also redirected to / + return {norm_path(p) for p in paths} + + +def classify_links(cfg, links, generated, media_urls, wp_ids, redirect_keys, media_dir): + """Split links into (fine, missing media, to probe, odd) lists.""" + fine, missing, probe, odd = [], [], [], [] + for link in sorted(links): + path, _, query = link.partition("?") + if query: + q = urllib.parse.parse_qs(query) + ident = next((q[k][0] for k in ("p", "page_id") if k in q), None) + if path == "/" and ident in wp_ids: + fine.append(link) + else: + odd.append(link) + continue + norm = norm_path(path) + if norm in generated or norm in redirect_keys or path in redirect_keys: + fine.append(link) + elif norm.startswith("/wp-content/uploads/"): + local = os.path.join(media_dir, urllib.parse.unquote(path.lstrip("/"))) + (fine if path in media_urls or os.path.exists(local) else missing).append(link) + elif norm.startswith("/wp-content/") or norm.startswith("/wp-"): + odd.append(link) + else: + probe.append(link) + return fine, missing, probe, odd + + +def probe_links(fetcher, wp, probe, generated, cache_file, refresh): + """HEAD every link; returns ({old: new} for 301s, [report lines]).""" + cache = {} + if os.path.exists(cache_file) and not refresh: + with open(cache_file, encoding="utf-8") as f: + cache = json.load(f) + redirects, lines = {}, [] + for link in probe: + if blog.htaccess_unsafe(link): + lines.append(f"{link!r}: skipped (a space or control character cannot go into .htaccess)") + continue + if link not in cache: + cache[link] = list(fetcher.head(wp.rstrip("/") + link)) + status, location = cache[link] + if status in (301, 308) and location: + target = urllib.parse.urlsplit(urllib.parse.urljoin(wp.rstrip("/") + link, location)) + if target.netloc == urllib.parse.urlsplit(wp).netloc: + new = (target.path or "/") + (f"?{target.query}" if target.query else "") + if blog.htaccess_unsafe(new): + lines.append(f"{link} -> {new!r}: skipped (a space or control character cannot go into .htaccess)") + continue + if norm_path(new) == norm_path(link): + lines.append(f"{link}: only adds a trailing slash or a query; handled by the generator") + continue + redirects[urllib.parse.quote(link, safe="/%:@!$&'*+,;=-._~")] = new + if norm_path(new) not in generated: + lines.append(f"{link} -> {new}: the target is not a generated page") + continue + lines.append(f"{link} -> {location}: leaves the site") + elif status == 200: + lines.append(f"{link}: exists on the old site but not on the new one") + else: + lines.append(f"{link}: HTTP {status}" + (f" ({location})" if location and status in (302, 307) else "")) + atomic_write(cache_file, json.dumps(cache, ensure_ascii=False, indent=1, sort_keys=True) + "\n") + return redirects, lines + + +# ---------------------------------------------------------------- applying + +def inside(root, path): + root, path = os.path.realpath(root), os.path.realpath(path) + return path == root or path.startswith(root + os.sep) + + +def write_file(src, rel, text, dry, force): + """'create', 'same', 'keep' (exists and differs: left alone) or 'overwrite'. Writes unless `dry`.""" + path = os.path.join(src, rel) + if not inside(src, os.path.dirname(path)): + raise WpImportError(f"refusing to write outside the source folder: {rel}") + if os.path.exists(path): + with open(path, encoding="utf-8") as f: + if f.read() == text: + return "same" + if not force: + return "keep" + action = "overwrite" + else: + action = "create" + if not dry: + atomic_write(path, text) + return action + + +def merge_dict(target, new, force, changes, label): + """Adds the missing keys of `new` to `target`; an existing value is replaced only with `force`.""" + for key, value in new.items(): + if key not in target: + target[key] = value + changes.append(f"{label}: add {key}") + elif target[key] != value: + if force: + target[key] = value + changes.append(f"{label}: replace {key}") + else: + changes.append(f"{label}: kept existing {key} (differs; use --force to replace)") + return target + + +def merge_config(raw, cfg, plan, opts, wp): + """(new raw config dict, [change lines]). Never overwrites without opts['force'].""" + force, changes = opts["force"], [] + new = json.loads(json.dumps(raw)) + include = list(new.get("include") or cfg["include"]) + if not any(p.startswith("**") or p == "wordpress/*.md" for p in include): + include.append("wordpress/*.md") + if include != new.get("include"): + new["include"] = include + changes.append("include: add wordpress/*.md") + has_pages = any(not rel.startswith("wordpress/") for rel in plan.files) + for key, value, wanted in (("pages_dir", "pages", has_pages), + ("media", "media", not opts["no_media"]), ("comments", "comments.json", bool(plan.comments)), + ("timezone", opts["timezone"], True), ("home", opts.get("home"), bool(opts.get("home")))): + if wanted and value and not new.get(key): + new[key] = value + changes.append(f"{key}: set to {value}") + hosts = list(new.get("rewrite_hosts") or []) + base = wp.rstrip("/") + for variant in (base, base.replace("https://", "http://", 1)): + if variant not in hosts: + hosts.append(variant) + changes.append(f"rewrite_hosts: add {variant}") + if hosts: + new["rewrite_hosts"] = hosts + for key, value in (("tags", plan.tags), ("wp_ids", plan.wp_ids), ("redirects", plan.redirects)): + if value: + new[key] = merge_dict(dict(new.get(key) or {}), dict(sorted(value.items())), force, changes, key) + posts = dict(new.get("posts") or {}) + for rel, entry in sorted(plan.post_entries.items()): + existing = dict(posts.get(rel) or {}) + posts[rel] = merge_dict(existing, entry, force, changes, f"posts[{rel}]") + new["posts"] = posts + return new, changes + + +def merge_comments(base, plan, comments_name, dry, force): + """`base` is the config file's folder: `comments` is relative to it, as the generator reads it.""" + path = os.path.join(base, comments_name) + existing = {} + if os.path.exists(path): + with open(path, encoding="utf-8") as f: + existing = json.load(f) + merged = {k: list(v) for k, v in existing.items()} + for key, lst in plan.comments.items(): + if force or key not in merged: + merged[key] = lst + else: + have = {c.get("id") for c in merged[key]} + merged[key] += [c for c in lst if c["id"] not in have] + text = json.dumps(merged, ensure_ascii=False, indent=1, sort_keys=True) + "\n" + same = False + if os.path.exists(path): + with open(path, encoding="utf-8") as f: + same = f.read() == text + if not same and not dry: + atomic_write(path, text) + return "same" if same else ("update" if existing else "create") + + +def media_patterns(): + pats = ["--include=*/"] + for ext in sorted(blog.MEDIA_EXTENSIONS): + pats.append("--include=*" + "".join(f"[{c.lower()}{c.upper()}]" if c.isalpha() else c for c in ext)) + return pats + ["--exclude=*"] + + +def sync_media(src, media_folder, source_spec, run=subprocess.run): + """Read-only rsync of the old uploads folder into <source>/<media>/wp-content/uploads/ (media types only).""" + dest = os.path.join(src, media_folder, "wp-content", "uploads") + if not inside(src, dest): + raise WpImportError("the media folder must be inside the source folder") + if not shutil.which("rsync"): + raise WpImportError("rsync is not installed") + os.makedirs(dest, exist_ok=True) + cmd = ["rsync", "-rt", "--prune-empty-dirs", *media_patterns(), source_spec.rstrip("/") + "/", dest + "/"] + result = run(cmd, capture_output=True, text=True) + if result.returncode: + raise WpImportError(f"rsync failed ({result.returncode}): {(result.stderr or '').strip()[:300]}") + return dest + + +# ---------------------------------------------------------------- the import + +def run_import(config_path, wp, dry_run=False, force=False, refresh=False, probe=False, cache=None, no_media=False, + media_from=None, timezone="America/Los_Angeles", home=None, fetcher=None, rsync=subprocess.run, + say=print): + """Runs the whole import; returns the report text. The report file wordpress/_report.txt holds only what + does not change between two runs (so a second run leaves it as it was); what this run did is added to the + returned text.""" + cfg = blog.load_config(config_path) + src = cfg["_source"] + wp = wp.rstrip("/") + wpdir = os.path.join(src, "wordpress") + api = os.path.abspath(cache) if cache else os.path.join(wpdir, "_api") + fetcher = fetcher or Fetcher(wp, api, refresh=refresh, log=say) + notes = [] + try: + zoneinfo.ZoneInfo(timezone) + except (zoneinfo.ZoneInfoNotFoundError, ValueError, OSError): + raise WpImportError(f"--timezone {timezone!r} is not an IANA time zone name") + + posts, skipped, ignored = read_posts(fetcher, notes) + pages, skipped_pages = read_pages(fetcher) + comments = fetcher.collection("comments") + try: + media = fetcher.collection("media") + except WpImportError as e: + say(f"media list not available: {e}") + media = [] + media_urls = set() + for m in media: + for url in [m.get("source_url")] + [s.get("source_url") for s in ((m.get("media_details") or {}).get("sizes") or {}).values()]: + if url: + media_urls.add(link_path(url, wp)) + + files = source_candidates(cfg) + assigned, best, conflicts = match_posts(posts, files) + if home is None and not cfg["home"] and any(p["slug"] == "highlights" for p in pages): + home = "highlights" + home = cfg["home"] or home + plan = build_plan(cfg, posts, pages, comments, assigned, files, notes, home) + + protected_ids = {str(s["id"]) for s in skipped} + skipped_comments = sum(1 for c in comments if str(c.get("post")) in protected_ids) + kept_comments = sum(len(v) for v in plan.comments.values()) + other_comments = len(comments) - kept_comments - skipped_comments + + # Links the new site will not generate. + hosts = [wp, wp.replace("https://", "http://", 1)] + [h.rstrip("/") for h in cfg["rewrite_hosts"]] + texts = [p["html"] for p in posts] + [pg["html"] for pg in pages] + texts += [blog.extract_body(files[j]["post"].text, cfg["body_section"], files[j]["post"].body_until, files[j]["rel"]) for _, j in assigned.values()] + links = set() + for t in texts: + links |= internal_links(t, hosts) + generated = generated_paths(cfg, posts, pages, plan, home) + media_dir = os.path.join(src, cfg["media"] or "media") + fine, missing_media, to_probe, odd = classify_links(cfg, links, generated, media_urls, plan.wp_ids, + set(blog.redirect_lookup(cfg)), media_dir) + probe_lines = [] + if probe and to_probe: + plan.redirects, probe_lines = probe_links(fetcher, wp, to_probe, generated, os.path.join(api, "probes.json"), refresh) + + # Apply (or only plan, in a dry run). + results = {} + for rel, text in sorted(plan.files.items()): + results[rel] = write_file(src, rel, text, dry_run, force) + config_changes = [] + with open(os.path.abspath(config_path), encoding="utf-8") as f: + raw = json.load(f) + opts = {"force": force, "no_media": no_media, "timezone": timezone, "home": home} + new_raw, config_changes = merge_config(raw, cfg, plan, opts, wp) + comments_name = new_raw.get("comments") or "comments.json" + comments_result = merge_comments(cfg["_dir"], plan, comments_name, dry_run, force) if plan.comments else "none" + media_result = "skipped (--no-media)" if no_media else "not run (dry run)" + if not dry_run: + if new_raw != raw: + atomic_write(os.path.abspath(config_path), json.dumps(new_raw, ensure_ascii=False, indent=2) + "\n") + blog.load_config(config_path) # the merged config must still be valid + if not no_media: + source_spec = media_from or f"dh:{urllib.parse.urlsplit(wp).hostname}/wp-content/uploads/" + try: + media_result = "copied from " + source_spec + " into " + os.path.relpath( + sync_media(src, new_raw.get("media") or "media", source_spec, rsync), src) + except WpImportError as e: + media_result = f"FAILED: {e}" + + report, actions = make_report(cfg, wp, dry_run, posts, skipped, ignored, pages, skipped_pages, comments, kept_comments, + skipped_comments, other_comments, media, files, assigned, best, conflicts, plan, results, + config_changes, comments_result, media_result, fine, missing_media, to_probe, odd, probe, + probe_lines, notes, timezone) + atomic_write(os.path.join(wpdir, "_report.txt"), report) + return report + actions + + +def section(lines, title, items): + lines.append(f"\n== {title} ==") + lines += [f" {i}" for i in items] if items else [" (none)"] + + +def make_report(cfg, wp, dry_run, posts, skipped, ignored, pages, skipped_pages, comments, kept_comments, skipped_comments, + other_comments, media, files, assigned, best, conflicts, plan, results, config_changes, comments_result, + media_result, fine, missing_media, to_probe, odd, probe, probe_lines, notes, timezone): + """(report file text, what-this-run-did text).""" + lines = ["WordPress import report", f"site: {wp}", + "", f"public posts: {len(posts)} password-protected (skipped): {len(skipped)} other status (ignored): {len(ignored)}", + f"pages: {len(pages)} comments: {len(comments)} (kept {kept_comments}, on skipped posts {skipped_comments}, other {other_comments})", + f"media library entries: {len(media)} source files considered: {len(files)}"] + section(lines, "Skipped: password-protected posts (their comments are skipped too)", [f'"{s["title"]}"' for s in skipped] + + [f'page "{t}"' for t in skipped_pages] + [f'{t} (ignored)' for t in ignored]) + matched = sorted(((posts[i]["path"], score, files[j]["rel"], posts[i]["title"]) for i, (score, j) in assigned.items())) + section(lines, f"Matched to a source file (score >= {MATCH_THRESHOLD}): {len(matched)}", + [f"{score:.4f} {rel} <- \"{title}\" {path}" for path, score, rel, title in matched]) + unmatched = [] + for i, post in enumerate(posts): + if i not in assigned: + cand = best.get(i) or [] + hint = f"best candidate {files[cand[0][1]]['rel']} at {cand[0][0]:.3f}" if cand else "no candidate" + unmatched.append(f'"{post["title"]}" ({post["date"].date()}) -> {post["rel"]} [{hint}]') + section(lines, f"Not matched (written from the WordPress HTML): {len(unmatched)}", unmatched) + section(lines, "Conflicts (double claims)", conflicts) + section(lines, "Files kept because they differ from what the importer would write (--force overwrites)", + [rel for rel, action in sorted(results.items()) if action == "keep"] + [c for c in config_changes if "kept existing" in c]) + section(lines, "Titles that differ from the source file (the config gets `title`)", + [f'{rel}: "{e["title"]}"' for rel, e in sorted(plan.post_entries.items()) if "title" in e and not rel.startswith("wordpress/")]) + mismatch = [p for p in posts if p.get("date_gmt") and not tz_agrees(p, timezone)] + section(lines, f"Dates that disagree with timezone {timezone} (check `timezone` in the config)", + [f'"{p["title"]}" {p["date"]} vs gmt {p["date_gmt"]}' for p in mismatch[:20]]) + links_info = [f"{len(fine)} internal link(s) fine (generated pages, redirects, media)"] + links_info += [f"missing media: {l}" for l in missing_media] + links_info += [f"unusual link (not handled): {l}" for l in odd] + if probe: + links_info += [f"{len(to_probe)} link(s) probed"] + probe_lines + links_info += [f"redirect {a} -> {b}" for a, b in sorted(plan.redirects.items())] + else: + links_info += [f"unchecked (use --probe-links): {l}" for l in to_probe] + section(lines, "Internal links", links_info) + section(lines, "Conversion notes (embeds dropped etc.)", sorted(set(notes))) + done = [f"\n== This run: {'DRY RUN (nothing written except _api/ and the report)' if dry_run else 'import'} =="] + done.append(f" files: {dict(collections.Counter(results.values())) or 'nothing to write'}") + done += [f" {action:9} {rel}" for rel, action in sorted(results.items()) if action != "same"] + done += [f" comments file: {comments_result}", f" media: {media_result}"] + done += [f" config: {c}" for c in config_changes] or [" config: no changes"] + return "\n".join(lines) + "\n", "\n".join(done) + "\n" + + +def tz_agrees(post, timezone): + try: + gmt = datetime.datetime.fromisoformat(post["date_gmt"]).replace(tzinfo=datetime.timezone.utc) + local = post["date"].replace(tzinfo=zoneinfo.ZoneInfo(timezone)) + return abs((gmt - local).total_seconds()) < 1 + except (ValueError, KeyError): + return True + + +# ---------------------------------------------------------------- check + +BODY_RE = re.compile(r'<div class="post-body">(.*?)</div></article>', re.S) + + +def output_has(out, path): + """True when the built site has a page or file at this URL path (a feed folder counts).""" + p = urllib.parse.unquote(path.split("?")[0].split("#")[0]).lstrip("/") + full = os.path.join(out, p) + if not inside(out, full): + return False + if os.path.isfile(full): + return True + return any(os.path.isfile(os.path.join(full, name)) for name in ("index.html", "index.xml")) + + +def run_check(config_path, out=None, cache=None, say=print): + """Compares the built site with the cached WordPress data. Returns (report text, number of problems).""" + cfg = blog.load_config(config_path) + src = cfg["_source"] + api = os.path.abspath(cache) if cache else os.path.join(src, "wordpress", "_api") + fetcher = Fetcher("http://invalid.invalid", api) + for name in ("posts", "pages", "tags"): + if fetcher.cached(name) is None: + raise WpImportError(f"no cached WordPress data in {api} (run fw blog import-wp --dry-run first)") + tmp = None + if out is None: + tmp = tempfile.mkdtemp(prefix="fw-blog-check-") + out = tmp + blog.build(cfg, out) + try: + return check_site(cfg, out, fetcher) + finally: + if tmp: + shutil.rmtree(tmp, ignore_errors=True) + + +def check_site(cfg, out, fetcher): + notes = [] + posts, skipped, _ = read_posts(fetcher, notes) + pages, _ = read_pages(fetcher) + problems, lines = 0, ["Blog check", f"output: {out}", f"posts compared: {len(posts)} (line_breaks = {cfg['line_breaks']})"] + low, brs, missing = [], [], [] + for p in posts: + page = os.path.join(out, urllib.parse.unquote(p["path"]).strip("/"), "index.html") + if not os.path.isfile(page): + missing.append(f'post {p["path"]} "{p["title"]}"') + continue + with open(page, encoding="utf-8") as f: + m = BODY_RE.search(f.read()) + built = m.group(1) if m else "" + score = similarity(p["words"], norm_words(blog.plain_text(built))) + if score < CHECK_THRESHOLD: + low.append((score, p["path"], p["title"])) + wp_br, built_br = html_br_count(p["html"]), len(re.findall(r"<br\b", built)) + if wp_br != built_br: + brs.append(f'{p["path"]} WordPress {wp_br}, built {built_br} "{p["title"]}"') + for p in pages: + want = "/" if p["path"] in ("/", f'/{cfg["home"]}/') else p["path"] + if p["slug"] != "blog" and not output_has(out, want): + missing.append(f'page {p["path"]} "{p["title"]}"') + tag_links = {} + for t in fetcher.cached("tags") or []: + tag_links[t["slug"]] = link_path(t.get("link"), fetcher.wp) + used = {t for p in posts for t in p["tags"]} + for slug in sorted(used): + if not output_has(out, tag_links.get(slug, f"/tag/{slug}/")): + missing.append(f"tag {tag_links.get(slug, f'/tag/{slug}/')}") + bad_targets = [] + targets = list(cfg["redirects"].items()) + [(f"?p={k}", v) for k, v in cfg["wp_ids"].items()] + for old, new in sorted(targets): + if not (re.match(r"[a-z]+://", new) or output_has(out, new)): + bad_targets.append(f"{old} -> {new}") + section(lines, f"Posts below {CHECK_THRESHOLD} similarity to the WordPress text: {len(low)}", + [f'{s:.3f} {path} "{title}"' for s, path, title in sorted(low)]) + section(lines, f"<br> counts that differ: {len(brs)}", brs) + section(lines, f"Old URLs missing from the output: {len(missing)}", missing) + section(lines, f"Redirect targets missing from the output: {len(bad_targets)}", bad_targets) + problems = len(low) + len(brs) + len(missing) + len(bad_targets) + lines.append(f"\n{'OK' if not problems else str(problems) + ' problem(s)'}") + return "\n".join(lines) + "\n", problems + + +# ---------------------------------------------------------------- command line + +def main(argv): + ap = argparse.ArgumentParser(prog="wp_import.py", description=__doc__.split("\n")[0]) + sub = ap.add_subparsers(dest="cmd", required=True) + i = sub.add_parser("import") + i.add_argument("config") + i.add_argument("--wp", required=True, help="the WordPress site, e.g. https://blog.example.org") + i.add_argument("--dry-run", action="store_true") + i.add_argument("--force", action="store_true", help="overwrite existing files and config entries") + i.add_argument("--refresh", action="store_true", help="fetch the API data again instead of using the cache") + i.add_argument("--probe-links", action="store_true", help="HEAD internal links the new site lacks; record 301 targets") + i.add_argument("--cache", help="folder for the API JSON (default <source>/wordpress/_api)") + i.add_argument("--no-media", action="store_true") + i.add_argument("--media-from", help="rsync source of the old uploads (default dh:<host>/wp-content/uploads/)") + i.add_argument("--timezone", default="America/Los_Angeles", help="written to the config when it has none") + i.add_argument("--home", help="slug of the WordPress page that is the front page (default: highlights, if present)") + c = sub.add_parser("check") + c.add_argument("config") + c.add_argument("--out", help="an already built site (default: build into a temporary folder)") + c.add_argument("--cache") + args = ap.parse_args(argv[1:]) + try: + if args.cmd == "import": + report = run_import(args.config, args.wp, dry_run=args.dry_run, force=args.force, refresh=args.refresh, + probe=args.probe_links, cache=args.cache, no_media=args.no_media, + media_from=args.media_from, timezone=args.timezone, home=args.home, + say=lambda s: print(s, file=sys.stderr)) + print(report) + return 1 if "FAILED" in report else 0 + report, problems = run_check(args.config, args.out, args.cache) + print(report) + return 1 if problems else 0 + except blog.BlogError as e: + print(f"wp_import.py: {e}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main(sys.argv)) diff --git a/server/atom.py b/server/atom.py new file mode 100644 index 0000000..42a524c --- /dev/null +++ b/server/atom.py @@ -0,0 +1,72 @@ +"""Atom 1.0 feeds, standard library only. + + atom_feed(title, url, updated, entries, rights, author=None, feed_url=None, subtitle=None) -> str + +Used by blog.py for the blog's /feed/. Each entry is a dict: + title, url, updated required (`updated` is a datetime, a date or an RFC 3339 string) + published, content, summary, id, author optional (`content` is HTML) +The XML is built with ElementTree, so every text is escaped by the library. +""" + +import datetime +import re +import xml.etree.ElementTree as ET + +NS = "http://www.w3.org/2005/Atom" +ET.register_namespace("", NS) + +# Characters XML 1.0 cannot carry at all (a stray form feed in a post would make the feed unparseable). +_ILLEGAL = re.compile("[\x00-\x08\x0b\x0c\x0e-\x1f]") + + +def rfc3339(value): + """2018-03-02T00:00:00Z. Naive datetimes and dates are written as UTC.""" + if isinstance(value, str): + return value + if isinstance(value, datetime.datetime): + if value.tzinfo: + value = value.astimezone(datetime.timezone.utc).replace(tzinfo=None) + return value.strftime("%Y-%m-%dT%H:%M:%SZ") + return datetime.datetime(value.year, value.month, value.day).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _el(parent, tag, text=None, **attrs): + e = ET.SubElement(parent, f"{{{NS}}}{tag}", {k: v for k, v in attrs.items() if v is not None}) + if text is not None: + e.text = _ILLEGAL.sub("", text) + return e + + +def _author(parent, name): + _el(_el(parent, "author"), "name", name) + + +def atom_feed(title, url, updated, entries, rights="", author=None, feed_url=None, subtitle=None): + root = ET.Element(f"{{{NS}}}feed") + _el(root, "title", title) + if subtitle: + _el(root, "subtitle", subtitle) + _el(root, "id", url) + _el(root, "updated", rfc3339(updated)) + _el(root, "link", rel="alternate", type="text/html", href=url) + _el(root, "link", rel="self", type="application/atom+xml", href=feed_url or url) + if rights: + _el(root, "rights", rights) + if author: + _author(root, author) + for e in entries: + item = _el(root, "entry") + _el(item, "title", e["title"]) + _el(item, "id", e.get("id") or e["url"]) + _el(item, "link", rel="alternate", type="text/html", href=e["url"]) + _el(item, "updated", rfc3339(e["updated"])) + if e.get("published"): + _el(item, "published", rfc3339(e["published"])) + if e.get("author"): + _author(item, e["author"]) + if e.get("summary"): + _el(item, "summary", e["summary"], type="text") + if e.get("content"): + _el(item, "content", e["content"], type="html") + ET.indent(root) + return '<?xml version="1.0" encoding="utf-8"?>\n' + ET.tostring(root, encoding="unicode") + "\n" diff --git a/server/blog.py b/server/blog.py new file mode 100644 index 0000000..be2e08d --- /dev/null +++ b/server/blog.py @@ -0,0 +1,954 @@ +#!/usr/bin/env python3 +"""Static blog generator (writing.fuzzyworld.net). Runs on the laptop, from Markdown files. + + blog.py list <config> every source file: path -> URL, title, date, tags (and whether it is published) + blog.py build <config> --out DIR write the site into DIR (--no-htaccess for a local preview) + blog.py deploy-target <config> print host:dir from the config (used by `fw blog deploy`) + blog.py audit DIR refuse (exit 1) if DIR holds anything that must not be uploaded + +`fw blog ...` is the usual way in (it also installs Markdown and Pygments into ~/fw/pylib). The config is +a JSON file next to the posts; its keys are listed in CONFIG_KEYS. `source` and `comments` are relative to +the config file (so is the importer's `comments.json`); `pages_dir` and `media` are relative to `source`. + +`timezone` (an IANA name, default UTC) says where post dates were written: feed timestamps use it. + +A post is a Markdown file `YYYY-MM-DD-name.md`. Optional front matter (a `---` block of `key: value` lines) +and the config's `posts` table can set published, date, slug, title, tags, body_until and wp_id; front matter +wins over the config, which wins over what the file name and its first `# ` heading say. A post is published +only when `published` is exactly true. Text under `## Spelling` and `## Removed writing`, and every HTML comment +except <!--more-->, never reaches the output. A heading that looks like a private one but is not written as +`## Spelling` (no space, an underlined heading, an unclosed code fence) stops the build instead of leaking. + +A page (a file in `pages_dir`) is published only with `published: true` in its front matter; it may set slug and title. + +Only the standard library is required, plus Python-Markdown (and Pygments for colored code) from ~/fw/pylib. +""" + +import argparse +import calendar +import dataclasses +import datetime +import glob +import html +import json +import math +import os +import re +import shutil +import sys +import urllib.parse +import zoneinfo + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) +import atom # noqa: E402 +import gen # noqa: E402 (also puts ~/fw/pylib, or $FW_PYLIB, on sys.path) + +markdown = gen.markdown +esc, write = gen.esc, gen.write + +SITE_DIR = os.path.join(HERE, "site") # _shared.css, favicon.ico: shared with code.fuzzyworld.net +BLOG_DIR = os.path.join(HERE, "site-blog") # blog.css, blog.js, htaccess, robots.txt +MARKER = ".fw-blog-out" # in an output folder this tool made: the only kind it empties +# Media are copied by an allowlist of extensions: ~/Blog holds much that must never be published (Markdown, +# JSON, notes, dotfiles). The audit applies the same rule again to everything in the build before an upload. +MEDIA_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".pdf", ".epub", ".kpf", ".mp3", ".mp4", ".zip"} +GENERATED_EXTENSIONS = {".html", ".xml", ".css", ".js", ".ico"} # what the generator itself writes +GENERATED_NAMES = {"robots.txt", ".htaccess"} # the only other files allowed in an upload +PRIVATE_SECTIONS = {"spelling", "removed writing"} # never published, wherever they are +SLUG_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._%-]*$") # every slug that becomes part of a path or a URL +MORE = "<!--more-->" +DEFAULT_URL = "/{yyyy}/{mm}/{dd}/{slug}/" +POST_FIELDS = ("published", "date", "slug", "title", "tags", "body_until", "wp_id") +PAGE_FIELDS = ("published", "slug", "title") +CONFIG_KEYS = {"title", "tagline", "base_url", "author", "copyright", "source", "include", "body_section", + "line_breaks", "url", "home", "pages_dir", "nav", "tags", "feed", "reading_wpm", "media", + "rewrite_hosts", "redirects", "wp_ids", "comments", "deploy", "posts", "timezone"} +REQUIRED_KEYS = ("title", "base_url", "source") +DEFAULTS = {"tagline": "", "author": "", "copyright": "© {years} {author}. All rights reserved.", + "include": ["*.md"], "body_section": "Writing", "line_breaks": "markdown", "url": DEFAULT_URL, + "home": None, "pages_dir": None, "nav": [], "tags": {}, "feed": {"path": "/feed/", "entries": 20}, + "reading_wpm": 200, "media": None, "rewrite_hosts": [], "redirects": {}, "wp_ids": {}, + "comments": None, "deploy": None, "posts": {}, "timezone": "UTC"} + + +class BlogError(Exception): + """A problem with the config or a source file, reported without a traceback.""" + + +# ---------------------------------------------------------------- config + +def load_config(path): + """The config as a dict with defaults filled in, plus `_dir` (the config's folder) and `_source` (absolute).""" + path = os.path.abspath(path) + try: + with open(path, encoding="utf-8") as f: + raw = json.load(f) + except OSError as e: + raise BlogError(f"cannot read config {path}: {e.strerror}") + except ValueError as e: + raise BlogError(f"config {path} is not valid JSON: {e}") + if not isinstance(raw, dict): + raise BlogError("config must be a JSON object") + unknown = sorted(set(raw) - CONFIG_KEYS) + if unknown: + raise BlogError("unknown config key(s): " + ", ".join(unknown)) + missing = [k for k in REQUIRED_KEYS if not raw.get(k)] + if missing: + raise BlogError("config needs: " + ", ".join(missing)) + cfg = {k: (dict(v) if isinstance(v, dict) else list(v) if isinstance(v, list) else v) for k, v in DEFAULTS.items()} + cfg.update(raw) + cfg["feed"] = {**DEFAULTS["feed"], **(cfg["feed"] or {})} + cfg["base_url"] = cfg["base_url"].rstrip("/") + if cfg["line_breaks"] not in ("br", "markdown"): + raise BlogError('line_breaks must be "br" or "markdown"') + try: + zoneinfo.ZoneInfo(str(cfg["timezone"])) + except (zoneinfo.ZoneInfoNotFoundError, ValueError, OSError): + raise BlogError(f'timezone {cfg["timezone"]!r} is not an IANA time zone name (e.g. "America/Los_Angeles")') + if not isinstance(cfg["posts"], dict) or not all(isinstance(v, dict) for v in cfg["posts"].values()): + raise BlogError("posts must map a file path to an object") + for rel, entry in cfg["posts"].items(): + bad = sorted(set(entry) - set(POST_FIELDS)) + if bad: + raise BlogError(f"posts[{rel}]: unknown key(s): " + ", ".join(bad)) + if not (isinstance(cfg["nav"], list) and all(isinstance(n, list) and len(n) == 2 for n in cfg["nav"])): + raise BlogError("nav must be a list of [label, path] pairs") + if not (cfg["url"].startswith("/") and cfg["url"].endswith("/") and "{slug}" in cfg["url"]): + raise BlogError("url must start and end with / and contain {slug}") + for old, new in cfg["redirects"].items(): + if (not isinstance(new, str) or htaccess_unsafe(old) or htaccess_unsafe(new) or old.strip("/") == ""): + raise BlogError(f"redirects: bad entry {old!r} -> {new!r} (no spaces or control characters, even percent-encoded)") + for key, value in cfg["wp_ids"].items(): + if not isinstance(value, str) or htaccess_unsafe(key) or htaccess_unsafe(value): + raise BlogError(f"wp_ids: bad entry {key!r} -> {value!r} (no spaces or control characters, even percent-encoded)") + if cfg["home"]: + if not isinstance(cfg["home"], str): + raise BlogError("home must be a page slug") + check_slug(cfg["home"], "home") + feed_path = cfg["feed"]["path"] + segments = [x for x in str(feed_path).split("/") if x] + if not (isinstance(feed_path, str) and feed_path.startswith("/") and segments) or htaccess_unsafe(feed_path): + raise BlogError(f"feed.path {feed_path!r} must be a path like /feed/") + for x in segments: + check_slug(x, "feed.path") + if not isinstance(cfg["tags"], dict): + raise BlogError("tags must map a slug to a name") + for slug in cfg["tags"]: + check_slug(slug, "tags") + cfg["_dir"] = os.path.dirname(path) + cfg["_source"] = os.path.normpath(os.path.join(cfg["_dir"], cfg["source"])) + return cfg + + +def check_slug(slug, what): + """The slug, or BlogError: a slug becomes a folder name and part of a URL (and of .htaccess), so it is ASCII + letters, digits and . _ % - only, and it must not decode to `.`, `..` or something with a slash.""" + decoded = urllib.parse.unquote(str(slug)) + if (not SLUG_RE.match(str(slug)) or decoded in (".", "..") or re.search(r"[/\\\x00-\x1f\x7f]", decoded)): + raise BlogError(f"{what}: bad slug {slug!r} (use letters, digits and . _ % - only)") + return slug + + +def htaccess_unsafe(value): + """True when a value that goes into .htaccess has a space or control character, written out or percent-encoded.""" + text = str(value) + return any(c.isspace() or ord(c) < 32 or ord(c) == 127 for c in text + urllib.parse.unquote(text)) + + +_DEPLOY_HOST = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._@-]*$") +_DEPLOY_DIR = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*(/[A-Za-z0-9._-]+)*$") + + +def deploy_target(cfg): + """host:dir for rsync. The dir is relative to the login folder and must look like a site folder (e.g. + writing.fuzzyworld.net): rsync --delete runs against it, so `~`, `.`, `/`, `..` and the like are refused.""" + deploy = cfg["deploy"] or {} + host, folder = deploy.get("host"), deploy.get("dir") + if not (isinstance(host, str) and isinstance(folder, str) and host and folder): + raise BlogError("config needs deploy: {host, dir}") + folder = folder.rstrip("/") + if (not _DEPLOY_HOST.match(host) or not _DEPLOY_DIR.match(folder) + or any(part in (".", "..") for part in folder.split("/")) or ".." in folder): + raise BlogError(f"deploy.dir {deploy.get('dir')!r} (or host) is not safe: use a site folder such as " + "writing.fuzzyworld.net, relative to the login folder") + return f"{host}:{folder}" + + +# ---------------------------------------------------------------- source files + +_FM_LINE = re.compile(r"^([A-Za-z_][\w-]*):\s*(.*?)\s*$") + + +def parse_front_matter(text): + """(meta dict, rest of the text). A leading `---` block counts only if it closes and every line is `key: value`; + otherwise the `---` was a horizontal rule and there is no front matter.""" + lines = text.split("\n") + if not lines or lines[0].strip() != "---": + return {}, text + for end in range(1, len(lines)): + if lines[end].strip() == "---": + break + else: + return {}, text + meta = {} + for line in lines[1:end]: + if not line.strip(): + continue + m = _FM_LINE.match(line) + if not m: + return {}, text + key, value = m.groups() + if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'": + value = value[1:-1] + meta[key] = {"true": True, "false": False}.get(value, value) + return meta, "\n".join(lines[end + 1:]) + + +_FENCE = re.compile(r"^ {0,3}(`{3,}|~{3,})(.*)$") +_H2 = re.compile(r"^ {0,3}##(?!#)\s+(.*?)\s*#*\s*$") +_ATX = re.compile(r"^\s*#{1,6}(.*)$") +_UNDERLINE = re.compile(r"^\s*(=+|-+)\s*$") +_TEMPLATE_LINE = re.compile(r"^[ \t]*Created by template, on .*$", re.M) + + +class _Fences: + """Fenced code blocks as CommonMark has them: a fence is 3+ backticks or tildes (a backtick fence has no backtick + in its info string, so an inline ```span``` is not one); it ends at a line of the same character, at least as + long, with nothing after it. feed(line) is True for a line that belongs to a block, the fence lines included.""" + + def __init__(self): + self.mark = None # (character, length) while inside a block + + def feed(self, line): + m = _FENCE.match(line) + if self.mark is None: + if m and not (m.group(1)[0] == "`" and "`" in m.group(2)): + self.mark = (m.group(1)[0], len(m.group(1))) + return True + return False + if m and m.group(1)[0] == self.mark[0] and len(m.group(1)) >= self.mark[1] and not m.group(2).strip(): + self.mark = None + return True + + +def remove_comments(text, name=None): + """Cuts every HTML comment (private notes must not reach page source) except the <!--more--> marker. + A comment ends at its first `-->`, but a <!--more--> inside a comment belongs to the comment. A comment that + is never closed is cut to the end of the text (with a warning on stderr when `name` says which file).""" + out, i = [], 0 + while True: + start = text.find("<!--", i) + if start == -1: + out.append(text[i:]) + break + out.append(text[i:start]) + if text.startswith(MORE, start): + out.append(MORE) + i = start + len(MORE) + continue + j = start + 4 + while True: + end, more = text.find("-->", j), text.find(MORE, j) + if more != -1 and (end == -1 or more < end): + j = more + len(MORE) + continue + break + if end == -1: + if name: + print(f"blog.py: warning: {name}: a `<!--` is never closed; the text from it to the end is left out", + file=sys.stderr) + break + i = end + 3 + return "".join(out) + + +def parse_sections(text, name=None): + """(title, preamble lines, [(heading, lines)]) of comment-free text. The title is the first `# ` line + (not kept in any lines); each `## ` section's lines start with its heading line. Fenced code is never + mistaken for headings.""" + title, pre, sections, fences = None, [], [], _Fences() + for line in remove_comments(text, name).replace("\r\n", "\n").split("\n"): + if not fences.feed(line): + if title is None and line.startswith("# "): + title = line[2:].strip() + continue + h = _H2.match(line) + if h: + sections.append((h.group(1), [line])) + continue + (sections[-1][1] if sections else pre).append(line) + return title, pre, sections + + +def _private_name(text): + return text.strip(" \t#:*_`").casefold() in PRIVATE_SECTIONS + + +def check_private_headings(lines, name=None): + """Fail closed: the body about to be published must hold nothing that looks like a private heading written some + other way than `## Spelling` (`##Spelling`, `Spelling` over a line of dashes, `### Spelling` ...) and no + code fence that is never closed (it would hide what follows from the section parser).""" + fences, prev = _Fences(), None + for n, line in enumerate(lines, 1): + if fences.feed(line): + prev = None + continue + m = _ATX.match(line) + if (m and _private_name(m.group(1))) or (not m and prev is not None and _UNDERLINE.match(line) and _private_name(prev)): + raise BlogError(f"{name or 'text'}: line {n} of the publishable text looks like a private heading " + f"({(prev if not m else line).strip()!r}); write it as `## Spelling` / `## Removed writing` " + "on a line of its own, or reword it") + prev = line + if fences.mark: + raise BlogError(f"{name or 'text'}: a code fence is never closed; fix it so private sections are not hidden by it") + + +def title_of(text): + return parse_sections(text)[0] + + +def _collect(sections, until): + """Lines of the sections up to the one headed `until` (exclusive), skipping private ones.""" + out = [] + for heading, lines in sections: + if until and heading.casefold() == until.casefold(): + break + if heading.casefold() not in PRIVATE_SECTIONS: + out += lines + return out + + +def extract_body(text, section="Writing", until=None, name=None): + """The publishable Markdown of a source file. + + With a `## <section>` heading: that section, up to the next `## ` heading, or (when `until` names a heading) + through every section up to it. Without one (or with section=None): everything after the `# ` title line. + Private sections, HTML comments (except <!--more-->) and the "Created by template" line are always dropped. + BlogError (naming `name`) if what is left has a private-looking heading in another form, or an unclosed fence. + """ + _, pre, sections = parse_sections(text, name) + target = next((i for i, (h, _) in enumerate(sections) if section and h.casefold() == section.casefold()), None) + if target is not None: + lines = sections[target][1][1:] + (_collect(sections[target + 1:], until) if until else []) + else: + lines = pre + _collect(sections, until) + body = _TEMPLATE_LINE.sub("", "\n".join(lines)).strip("\n").strip() + check_private_headings(body.split("\n"), name) + return body + + +@dataclasses.dataclass +class Post: + rel: str # path under `source`, with / + text: str # the file without its front matter + published: bool = False + date: datetime.datetime = None + slug: str = "" + title: str = "" + tags: list = dataclasses.field(default_factory=list) + body_until: str = None + wp_id: int = None + path: str = "" # URL path, e.g. /2018/03/02/slug/ + + +def parse_date(value): + """datetime from "YYYY-MM-DD" or "YYYY-MM-DD[ T]HH:MM[:SS]". A time zone is dropped: dates are WordPress local dates.""" + if isinstance(value, datetime.datetime): + return value.replace(tzinfo=None) + try: + return datetime.datetime.fromisoformat(str(value).strip()).replace(tzinfo=None) + except ValueError: + raise BlogError(f"bad date {value!r} (use YYYY-MM-DD)") + + +def localize(cfg, when): + """A post's naive date as a time-aware datetime in the config's `timezone` (for feed timestamps).""" + return when.replace(tzinfo=zoneinfo.ZoneInfo(cfg["timezone"])) if when.tzinfo is None else when + + +def as_list(value): + if isinstance(value, str): + value = value.strip() + if value.startswith("[") and value.endswith("]"): + value = value[1:-1] + return [t for t in (t.strip().strip("\"'").strip() for t in value.split(",")) if t] + return [str(t).strip() for t in (value or []) if str(t).strip()] + + +_NAME = re.compile(r"^(\d{4}-\d{2}-\d{2})-(.+)\.md$", re.I) + + +def post_url(cfg, date, slug): + return (cfg["url"].replace("{yyyy}", f"{date.year:04d}").replace("{mm}", f"{date.month:02d}") + .replace("{dd}", f"{date.day:02d}").replace("{slug}", slug)) + + +def source_files(cfg, patterns=None, skip_dir=None): + """Relative paths of the source files matching the include patterns, sorted. Dropbox's "conflicted copy" + files and hidden folders are left out.""" + src, found = cfg["_source"], set() + for pattern in patterns or cfg["include"]: + for p in glob.glob(os.path.join(glob.escape(src), pattern), recursive=True): + rel = os.path.relpath(p, src).replace(os.sep, "/") + parts = rel.split("/") + if (not os.path.isfile(p) or "conflicted copy" in rel.lower() or any(x.startswith(".") for x in parts) + or (skip_dir and rel.startswith(skip_dir + "/"))): + continue + found.add(rel) + return sorted(found) + + +def read_text(path): + with open(path, encoding="utf-8-sig") as f: + return f.read().replace("\r\n", "\n") + + +def pages_dir(cfg): + return cfg["pages_dir"].strip("/") if cfg["pages_dir"] else None + + +def scan_posts(cfg): + """Every post file as a Post, oldest first. Precedence: front matter > config posts[<path>] > the file.""" + posts = [] + for rel in source_files(cfg, skip_dir=pages_dir(cfg)): + meta, text = parse_front_matter(read_text(os.path.join(cfg["_source"], rel))) + bad = sorted(set(meta) - set(POST_FIELDS)) + if bad: + raise BlogError(f"{rel}: unknown front matter key(s): " + ", ".join(bad)) + merged = {**cfg["posts"].get(rel, {}), **meta} + name = _NAME.match(os.path.basename(rel)) + date = merged.get("date") or (name.group(1) if name else None) + stem = name.group(2) if name else os.path.splitext(os.path.basename(rel))[0] + post = Post(rel=rel, text=text, + published=merged.get("published") is True, + date=parse_date(date) if date else None, + slug=str(merged.get("slug") or stem), + title=str(merged.get("title") or title_of(text) or stem), + tags=as_list(merged.get("tags")), + body_until=merged.get("body_until") or None, + wp_id=int(merged["wp_id"]) if str(merged.get("wp_id", "")).isdigit() else None) + if post.published: # a draft's file name may hold anything; it is never written anywhere + check_slug(post.slug, rel) + if post.date: + post.path = post_url(cfg, post.date, post.slug) + posts.append(post) + posts.sort(key=lambda p: (p.date or datetime.datetime.min, p.rel)) + return posts + + +def slugify(text): + return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") + + +def tag_slug(cfg, tag): + """A post's tag may be written as a configured slug, a configured name or any other text.""" + if tag in cfg["tags"]: + return tag + for slug, name in cfg["tags"].items(): + if name == tag: + return slug + return slugify(tag) or "tag" + + +def tag_name(cfg, slug, fallback=None): + return cfg["tags"].get(slug) or fallback or slug + + +def list_posts(cfg): + lines = [] + for p in scan_posts(cfg): + tags = ", ".join(p.tags) + lines.append(f'{"published" if p.published else "draft "} {p.rel} -> {p.path or "(no date)"} ' + f'{p.title!r} {p.date.date() if p.date else "no date"}' + (f" tags: {tags}" if tags else "")) + return lines + + +# ---------------------------------------------------------------- rendering + +def redirect_lookup(cfg): + """{path: target} from the config's redirects and wp_ids, tolerant of a missing or extra trailing slash.""" + if "_lookup" not in cfg: + table = {} + for old, new in cfg["redirects"].items(): + for key in (old, urllib.parse.unquote(old)): + table[key] = new + table[key.rstrip("/") + "/"] = new + table[key.rstrip("/")] = new + cfg["_lookup"] = table + return cfg["_lookup"] + + +def rewrite_url(url, cfg): + """Absolute links to the old host become site-relative; internal links go through `redirects` + (and WordPress's /?p=ID short links through `wp_ids`).""" + for prefix in cfg["rewrite_hosts"]: + p = prefix.rstrip("/") + if url == p or url[len(p):len(p) + 1] in ("/", "?", "#") and url.startswith(p): + url = url[len(p):] + url = url if url.startswith("/") else "/" + url + break + if not url.startswith("/") or url.startswith("//"): + return url + m = re.match(r"([^?#]*)(.*)$", url) + path, tail = m.group(1), m.group(2) + if path == "/": + q = urllib.parse.parse_qs(tail.lstrip("?").split("#")[0]) + for key in ("p", "page_id"): + target = cfg["wp_ids"].get((q.get(key) or [None])[0]) + if target: + return target + target = redirect_lookup(cfg).get(path) + if target: + return target if re.search(r"[?#]", target) else target + tail + return url + + +_ATTR = re.compile(r'(<(?:img|source)\b[^>]*?\bsrc|<a\b[^>]*?\bhref)=(["\'])(.*?)\2', re.I | re.S) + + +def map_links(html_text, fn): + def fix(m): + return f"{m.group(1)}={m.group(2)}{esc(fn(html.unescape(m.group(3))), quote=True)}{m.group(2)}" + return _ATTR.sub(fix, html_text) + + +_CODE = re.compile(r'<pre><code(?: class="language-([\w+#.-]+)")?>(.*?)</code></pre>', re.S) + + +def highlight_blocks(html_text): + """Fenced code blocks with a language get Pygments token spans (the classes gen.py uses).""" + def fix(m): + lang = m.group(1) + if not lang or not gen.pygments_lexers: + return m.group(0) + try: + lexer = gen.pygments_lexers.get_lexer_by_name(lang, stripnl=False, ensurenl=False) + parts = [] + for ttype, value in lexer.get_tokens(html.unescape(m.group(2))): + cls = gen.token_class(ttype) + parts.append(f'<span class="{cls}">{esc(value)}</span>' if cls else esc(value)) + except Exception: # unknown language or a lexer bug: leave the block plain + return m.group(0) + return f'<pre class="hl"><code class="language-{lang}">{"".join(parts)}</code></pre>' + return _CODE.sub(fix, html_text) + + +def render_markdown(text, cfg): + if not markdown: + raise BlogError("Python-Markdown is not installed in " + gen.PYLIB + " (fw blog installs it)") + extensions = ["tables", "fenced_code", "sane_lists"] + (["nl2br"] if cfg["line_breaks"] == "br" else []) + out = markdown.markdown(text, extensions=extensions, output_format="html") + return map_links(highlight_blocks(out), lambda u: rewrite_url(u, cfg)) + + +def plain_text(html_text): + return re.sub(r"\s+", " ", html.unescape(re.sub(r"<[^>]+>", " ", html_text))).strip() + + +def split_more(text): + """(excerpt Markdown or None, full Markdown) around the <!--more--> marker.""" + if "<!--more-->" not in text: + return None, text + head, tail = text.split("<!--more-->", 1) + return head.strip(), head.rstrip() + "\n\n" + tail.replace("<!--more-->", "").lstrip() + + +def blocks(html_text): + """The text blocks of rendered HTML as (kind, escaped-free text) pairs: kind is "p" or "h" (a chapter heading). + <br> is kept as a newline; other markup and images are dropped.""" + out = [] + for m in re.finditer(r"<(p|h[1-6])(?:\s[^>]*)?>(.*?)</\1>", html_text, re.S): + marked = re.sub(r"<br\s*/?>", "\x00", m.group(2)) + text = html.unescape(re.sub(r"<[^>]+>", "", marked)) + text = "\n".join(re.sub(r"[ \t]+", " ", line).strip() for line in text.split("\x00")).strip() + if text: + out.append(("p" if m.group(1) == "p" else "h", text)) + return out + + +def make_excerpt(excerpt_md, body_html, cfg, words=80): + """(HTML for the lists, whether there is more to read). What is before <!--more-->, else whole blocks from the + start until about `words` words (a first paragraph longer than that is cut at a word, even after headings). Chapter headings are + kept, but never left dangling at the end.""" + cut = excerpt_md is not None + found = blocks(render_markdown(excerpt_md, cfg) if cut else body_html) + if cut: + kept, more = found, True + else: + kept, n, more = [], 0, False + for kind, text in found: + w = len(text.split()) + if kind == "p" and w > words and not any(k == "p" for k, _ in kept): + text, more = " ".join(text.split(" ")[:words]) + "…", True # the first paragraph, headings or not + elif kept and n + w > words: + break + kept.append((kind, text)) + n += w + while kept and kept[-1][0] == "h" and len(kept) < len(found): + kept.pop() + more = more or len(kept) < len(found) + html_out = "".join(("<p>" if k == "p" else '<h5 class="chapter">') + esc(t).replace("\n", "<br>\n") + ("</p>" if k == "p" else "</h5>") + for k, t in kept) + return html_out, bool(more) + + +def absolutize(html_text, base_url): + """For the feed: site-relative links become absolute.""" + return map_links(html_text, lambda u: base_url + u if u.startswith("/") and not u.startswith("//") else u) + + +def render_comments(comments): + """Threaded comments: name, date and text only. Everything is escaped; newlines become <br>. Replies nest.""" + ids = {c.get("id") for c in comments} + children = {} + for c in sorted(comments, key=lambda c: str(c.get("date", ""))): + parent = c.get("parent") or 0 + children.setdefault(parent if parent in ids and parent != c.get("id") else 0, []).append(c) + + def tree(parent): + items = [] + for c in children.get(parent, []): + text = esc(str(c.get("text", "")).replace("\r\n", "\n")).replace("\n", "<br>\n") + nested = tree(c.get("id")) if c.get("id") in children else "" + items.append(f'<li class="comment" id="comment-{re.sub(r"[^A-Za-z0-9_-]", "", str(c.get("id", "")))}">' + f'<p class="meta"><strong>{esc(str(c.get("author", "")))}</strong> · ' + f'{esc(str(c.get("date", ""))[:10])}</p><p>{text}</p>{nested}</li>') + return f'<ul class="comments">{"".join(items)}</ul>' if items else "" + + if not comments: + return "" + return ('<section class="comments-section"><h2>Comments</h2>' + tree(0) + + '<p class="meta">Comments © their authors.</p></section>') + + +def load_comments(cfg): + if not cfg["comments"]: + return {} + path = os.path.join(cfg["_dir"], cfg["comments"]) + if not os.path.exists(path): + return {} + try: + with open(path, encoding="utf-8") as f: + data = json.load(f) + except ValueError as e: + raise BlogError(f"comments file {path} is not valid JSON: {e}") + return data if isinstance(data, dict) else {} + + +def copyright_text(cfg): + return (cfg["copyright"].replace("{years}", cfg.get("_years", str(datetime.date.today().year))) + .replace("{author}", cfg["author"])) + + +def theme_overrides(shared_css): + """CSS that lets a reader's choice (<html data-theme="light|dark">) beat the system setting. _shared.css holds the + light colors in :root and the dark ones in a prefers-color-scheme media query; both blocks are repeated here under + the attribute, taken from _shared.css itself so they cannot drift.""" + light = re.search(r":root\s*\{([^}]*--color-bg-primary[^}]*)\}", shared_css) + dark = re.search(r"@media \(prefers-color-scheme: dark\)\s*\{\s*:root\s*\{([^}]*)\}", shared_css) + if not (light and dark): + raise BlogError("_shared.css: cannot find the light and dark color blocks") + def decls(block): + return "\n".join(" " + line.strip() for line in block.splitlines() if line.strip().startswith("--")) + return (f'\n:root[data-theme="light"] {{\n color-scheme: light;\n{decls(light.group(1))}\n}}\n' + f':root[data-theme="dark"] {{\n color-scheme: dark;\n{decls(dark.group(1))}\n}}\n') + + +def blog_page(cfg, title, body, path): + """A full page. `path` is the page's own URL path (for the canonical link and the current nav entry).""" + links = "".join(f'<a href="{esc(p, quote=True)}"' + (' aria-current="page"' if p == path else "") + f'>{esc(label)}</a>' + for label, p in cfg["nav"]) + feed = cfg["feed"]["path"] + head_title = cfg["title"] if path == "/" else f'{title} · {cfg["title"]}' + return f"""<!doctype html> +<html lang="en"> +<head> +<meta charset="utf-8"> +<meta name="viewport" content="width=device-width, initial-scale=1"> +<title>{esc(head_title)}</title> +<link rel="canonical" href="{esc(cfg["base_url"] + path, quote=True)}"> +<link rel="alternate" type="application/atom+xml" title="{esc(cfg["title"], quote=True)}" href="{esc(feed, quote=True)}"> +<link rel="icon" href="/favicon.ico"> +<link rel="stylesheet" href="/_shared.css"> +<link rel="stylesheet" href="/blog.css"> +<script src="/blog.js"></script> +</head> +<body> +<header class="site"><div class="bar"><a class="brand" href="/">{esc(cfg["title"])}</a> +{f'<span class="tagline">{esc(cfg["tagline"])}</span>' if cfg["tagline"] else ""} +<nav class="site-nav" aria-label="Site">{links}</nav></div></header> +<main> +{body} +</main> +<footer class="site"><div class="cols"> +<p>{esc(copyright_text(cfg))}</p> +<p><a href="https://fuzzyworld.net/">fuzzyworld.net</a> · <a href="{esc(feed, quote=True)}">Feed</a></p> +</div></footer> +</body> +</html> +""" + + +def post_meta_line(cfg, post, minutes): + tags = "".join(f' <a class="tag" href="/tag/{esc(s, quote=True)}/">{esc(tag_name(cfg, s, t))}</a>' + for s, t in ((tag_slug(cfg, t), t) for t in post.tags)) + return (f'<p class="meta"><time datetime="{post.date.date()}">{post.date.date()}</time> · {minutes} min read' + + (" ·" + tags if tags else "") + "</p>") + + +def listing(cfg, items, headings=True): + """Posts (dicts with post, minutes, excerpt) as a list, newest first; with `headings`, grouped by year and month.""" + out, year, month = [], None, None + for it in items: + p = it["post"] + if headings and p.date.year != year: + year, month = p.date.year, None + out.append(f'<h2 id="{year}">{year}</h2>') + if headings and (p.date.year, p.date.month) != month: + month = (p.date.year, p.date.month) + out.append(f'<h3 id="{year}-{p.date.month:02d}">{calendar.month_name[p.date.month]} {year}</h3>') + out.append(f'<article class="entry"><h4><a href="{esc(p.path, quote=True)}">{esc(p.title)}</a></h4>' + f'{post_meta_line(cfg, p, it["minutes"])}' + + (f'<div class="excerpt">{it["excerpt"][0]}</div>' if it["excerpt"][0] else "") + + (f'<p class="more"><a href="{esc(p.path, quote=True)}">Continue reading →</a></p>' if it["excerpt"][1] else "") + + "</article>") + return "\n".join(out) + + +# ---------------------------------------------------------------- site files + +def apache_pattern(path): + """A RewriteRule pattern (per-directory context: no leading slash) matching exactly this path, with or without a trailing slash.""" + text = urllib.parse.unquote(path).strip("/") + text = re.sub(r"([.^$*+?()\[\]{}|\\])", r"\\\1", text).replace(" ", r"\x20") + return f"^{text}/?$" + + +def redirect_rules(cfg): + """Apache rules for the old WordPress URLs, from the config and the fixed WordPress URL shapes.""" + rules = ["", "# Redirects from the old WordPress site (generated by blog.py from the config)."] + for old, new in sorted(cfg["redirects"].items()): + rules.append(f"RewriteRule {apache_pattern(old)} {new} [R=301,L]") + for wp_id, path in sorted(cfg["wp_ids"].items(), key=lambda kv: str(kv[0])): + for param in ("p", "page_id"): + rules += [f"RewriteCond %{{QUERY_STRING}} (^|&){param}={re.escape(str(wp_id))}(&|$)", + f"RewriteRule ^$ {path}? [R=301,L]"] + if cfg["home"]: + rules.append(f"RewriteRule {apache_pattern(cfg['home'])} / [R=301,L]") + feed = cfg["feed"]["path"] + rules += [f"RewriteRule ^comments/feed/?$ {feed} [R=301,L]", + f"RewriteRule ^feed/(rss2|rss|rdf|atom)/?$ {feed} [R=301,L]", + "RewriteRule ^tag/([^/]+)/feed/?$ /tag/$1/ [R=301,L]"] + if cfg["url"] == DEFAULT_URL: + rules.append(r"RewriteRule ^(\d{4}/\d\d/\d\d/[^/]+)/(feed|amp|comment-page-\d+|attachment/.+)/?$ /$1/ [R=301,L]") + rules += ["RewriteRule ^(category|author)/.*$ /blog/ [R=301,L]", + r"RewriteRule ^page/\d+/?$ /blog/ [R=301,L]", + r"RewriteRule ^(\d{4})/?$ /blog/#$1 [NE,R=301,L]", + r"RewriteRule ^(\d{4})/(\d\d)/?$ /blog/#$1-$2 [NE,R=301,L]", + r"RewriteRule ^(wp-admin|wp-login\.php|xmlrpc\.php) - [R=404,L]"] + return "\n".join(rules) + "\n" + + +def htaccess_text(cfg): + with open(os.path.join(BLOG_DIR, "htaccess"), encoding="utf-8") as f: + return f.read().rstrip("\n") + "\n" + redirect_rules(cfg) + + +def copy_media(cfg, out): + """The media folder's files, as they are (so /wp-content/uploads/... paths keep working). Only file types + a browser shows or plays: nothing the web server could execute. Symbolic links (files or folders) are skipped.""" + if not cfg["media"]: + return 0 + root, n = os.path.join(cfg["_source"], cfg["media"]), 0 + for dirpath, dirs, files in os.walk(root): + dirs[:] = [d for d in dirs if not d.startswith(".") and not os.path.islink(os.path.join(dirpath, d))] + for name in files: + if (name.startswith(".") or os.path.splitext(name)[1].lower() not in MEDIA_EXTENSIONS + or os.path.islink(os.path.join(dirpath, name))): # a link could point at anything on the laptop + continue + dest = os.path.join(out, os.path.relpath(os.path.join(dirpath, name), root)) + os.makedirs(os.path.dirname(dest), exist_ok=True) + shutil.copyfile(os.path.join(dirpath, name), dest) + n += 1 + return n + + +def audit_output(out): + """(file count, total bytes) of what an upload of `out` would send; BlogError if anything must not be published. + Allowed: generated pages, feed, css, icon, robots.txt, .htaccess, and media with an allowlisted extension. + Anything else (a .md, .json or .txt file, any other dotfile, a symlink) is refused. The build's own marker + file is skipped: the upload excludes it.""" + if not os.path.isdir(out): + raise BlogError(f"{out} is not a folder") + bad, count, size = [], 0, 0 + for dirpath, dirs, files in os.walk(out): + for name in dirs: + if name.startswith(".") or os.path.islink(os.path.join(dirpath, name)): + bad.append(os.path.relpath(os.path.join(dirpath, name), out)) + for name in files: + path = os.path.join(dirpath, name) + if name == MARKER and dirpath == out: + continue + ext = os.path.splitext(name)[1].lower() + if (os.path.islink(path) or not (name in GENERATED_NAMES or (not name.startswith(".") + and ext in GENERATED_EXTENSIONS | MEDIA_EXTENSIONS))): + bad.append(os.path.relpath(path, out)) + continue + count += 1 + size += os.path.getsize(path) + if bad: + raise BlogError("refusing to upload; not allowed in the output: " + ", ".join(sorted(bad)[:20]) + + (f" (and {len(bad) - 20} more)" if len(bad) > 20 else "")) + return count, size + + +def safe_join(out, *parts): + """out/parts..., or BlogError when it would land outside `out` (a bad slug, a symlink in the way).""" + dest = os.path.join(out, *parts) + root, real = os.path.realpath(out), os.path.realpath(dest) + if not real.startswith(root + os.sep): + raise BlogError(f"refusing to write outside the output folder: {os.path.join(*parts)!r}") + return dest + + +def prepare_out(out, cfg=None): + """Empty the output folder, but only one this tool made (or an empty one): `--out ~` must not wipe a home folder.""" + if cfg: + out_real, src = os.path.realpath(out), os.path.realpath(cfg["_source"]) + if out_real == src or out_real.startswith(src + os.sep) or src.startswith(out_real + os.sep): + raise BlogError("--out must be a separate folder, not the source folder or inside it") + if os.path.exists(out): + if os.listdir(out) and not os.path.exists(os.path.join(out, MARKER)): + raise BlogError(f"{out} is not empty and was not made by fw blog; choose another --out") + shutil.rmtree(out) + os.makedirs(out) + write(os.path.join(out, MARKER), "Made by fw blog build. It is emptied on every build.\n") + + +# ---------------------------------------------------------------- build + +def build(cfg, out, htaccess=True): + """Writes the whole site into `out`. Returns a summary string.""" + posts = [p for p in scan_posts(cfg) if p.published] + for p in posts: + if not p.date: + raise BlogError(f"{p.rel}: published, but no date (name the file YYYY-MM-DD-name.md or set date)") + posts.sort(key=lambda p: (p.date, p.rel), reverse=True) + if posts: + first, last = posts[-1].date.year, posts[0].date.year + cfg["_years"] = str(first) if first == last else f"{first}–{last}" + comments = load_comments(cfg) + prepare_out(out, cfg) + written = set() + + def emit(path, content): + folder = urllib.parse.unquote(path).strip("/") # the web server decodes %xx before it looks for the file + if folder + path[-1:] in written: + raise BlogError(f"two pages want the URL {path}") + written.add(folder + path[-1:]) + write(safe_join(out, folder, "index.html") if path.endswith("/") else safe_join(out, folder), content) + + items, tags = [], {} + for p in posts: + excerpt_md, full = split_more(extract_body(p.text, cfg["body_section"], p.body_until, p.rel)) + body = render_markdown(full, cfg) + text = plain_text(body) + minutes = max(1, math.ceil(len(text.split()) / cfg["reading_wpm"])) + it = {"post": p, "html": body, "minutes": minutes, "excerpt": make_excerpt(excerpt_md, body, cfg)} + items.append(it) + for t in p.tags: + tags.setdefault(tag_slug(cfg, t), []).append(it) + page_body = (f'<article class="post"><h1>{esc(p.title)}</h1>{post_meta_line(cfg, p, minutes)}' + f'<div class="post-body">{body}</div></article>{render_comments(comments.get(p.path, []))}') + emit(p.path, blog_page(cfg, p.title, page_body, p.path)) + + years = sorted({it["post"].date.year for it in items}, reverse=True) + index = ('<h1>Blog</h1>\n' + ('<p class="years">' + " ".join(f'<a href="#{y}">{y}</a>' for y in years) + "</p>\n" + + listing(cfg, items) if items else "<p>No posts yet.</p>")) + emit("/blog/", blog_page(cfg, "Blog", index, "/blog/")) + for slug, its in sorted(tags.items()): + name = tag_name(cfg, slug, next((t for t in its[0]["post"].tags if tag_slug(cfg, t) == slug), slug)) + emit(f"/tag/{slug}/", blog_page(cfg, name, f'<h1>{esc(name)}</h1>\n{listing(cfg, its, headings=False)}', f"/tag/{slug}/")) + + home_done = False + pdir = pages_dir(cfg) + for rel in source_files(cfg, patterns=[pdir + "/*.md"]) if pdir else []: + meta, text = parse_front_matter(read_text(os.path.join(cfg["_source"], rel))) + bad = sorted(set(meta) - set(PAGE_FIELDS)) + if bad: + raise BlogError(f"{rel}: unknown front matter key(s): " + ", ".join(bad)) + if meta.get("published") is not True: # like posts: a page is published only when it says so + continue + stem = os.path.splitext(os.path.basename(rel))[0] + slug = check_slug(str(meta.get("slug") or stem), rel) + title = str(meta.get("title") or title_of(text) or stem) + page_html = render_markdown(extract_body(text, None, name=rel), cfg) + if slug == cfg["home"]: + emit("/", blog_page(cfg, title, f'<article class="post"><h1>{esc(title)}</h1>{page_html}</article>', "/")) + home_done = True + else: + path = f"/{slug}/" + emit(path, blog_page(cfg, title, f'<article class="post"><h1>{esc(title)}</h1>{page_html}</article>', path)) + if cfg["home"] and not home_done: + raise BlogError(f'home page "{cfg["home"]}" not found in pages_dir') + if not home_done: + emit("/", blog_page(cfg, cfg["title"], index, "/")) + emit("/404.html", blog_page(cfg, "Not found", '<h1>Not found</h1><p>There is nothing at this address. ' + '<a href="/blog/">All posts</a></p>', "/404.html")) + + entries = [{"title": it["post"].title, "url": cfg["base_url"] + it["post"].path, "updated": localize(cfg, it["post"].date), + "published": localize(cfg, it["post"].date), "content": absolutize(it["html"], cfg["base_url"])} + for it in items[:cfg["feed"]["entries"]]] + feed_path = cfg["feed"]["path"].strip("/") + write(safe_join(out, urllib.parse.unquote(feed_path), "index.xml"), + atom.atom_feed(cfg["title"], cfg["base_url"] + "/", + localize(cfg, posts[0].date) if posts else datetime.datetime.now(datetime.timezone.utc), + entries, copyright_text(cfg), author=cfg["author"] or None, + feed_url=f'{cfg["base_url"]}/{feed_path}/', subtitle=cfg["tagline"] or None)) + + for name in ("_shared.css", "favicon.ico"): + shutil.copyfile(os.path.join(SITE_DIR, name), os.path.join(out, name)) + for name in ("blog.js", "robots.txt"): + shutil.copyfile(os.path.join(BLOG_DIR, name), os.path.join(out, name)) + with open(os.path.join(SITE_DIR, "_shared.css")) as f: + shared = f.read() + with open(os.path.join(BLOG_DIR, "blog.css")) as f: + write(os.path.join(out, "blog.css"), f.read() + theme_overrides(shared)) + media = copy_media(cfg, out) + if htaccess: + write(os.path.join(out, ".htaccess"), htaccess_text(cfg)) + return f"{len(posts)} posts, {len(tags)} tags, {len(written)} pages, {media} media files -> {out}" + + +# ---------------------------------------------------------------- command line + +def main(argv): + ap = argparse.ArgumentParser(prog="blog.py", description=__doc__.split("\n")[0]) + sub = ap.add_subparsers(dest="cmd", required=True) + sub.add_parser("list").add_argument("config") + b = sub.add_parser("build") + b.add_argument("config") + b.add_argument("--out", required=True) + b.add_argument("--no-htaccess", action="store_true", help="skip .htaccess (local preview)") + sub.add_parser("audit").add_argument("out") + sub.add_parser("deploy-target").add_argument("config") + args = ap.parse_args(argv[1:]) + try: + if args.cmd == "audit": + count, size = audit_output(args.out) + print(f"audit ok: {count} files, {size / 1024:.0f} KB") + return + cfg = load_config(args.config) + if args.cmd == "list": + print("\n".join(list_posts(cfg))) + elif args.cmd == "build": + print(build(cfg, os.path.abspath(args.out), htaccess=not args.no_htaccess)) + else: + print(deploy_target(cfg)) + except BlogError as e: + raise SystemExit(f"blog.py: {e}") + + +if __name__ == "__main__": + main(sys.argv) diff --git a/server/site-blog/blog.css b/server/site-blog/blog.css new file mode 100644 index 0000000..390c4ba --- /dev/null +++ b/server/site-blog/blog.css @@ -0,0 +1,86 @@ +/* Styles for the blog (writing.fuzzyworld.net). Loaded after _shared.css, which supplies the color, + spacing and type tokens and the dark mode. A reading column, no frills. */ + +/* Same look as code.fuzzyworld.net (see site/style.css): a gray bar with the brand and an underlined current page, + a 72rem frame, left-aligned; running text is held to a reading width inside it. */ +body { margin: 0; font: 16px/1.65 var(--font-family-sans); } +a { color: var(--color-accent); text-decoration: none; } +a:hover { text-decoration: underline; } + +header.site { background: var(--color-bg-secondary); border-bottom: 1px solid var(--color-border-primary); } +header.site .bar, footer.site .cols, main { max-width: 72rem; margin: 0 auto; padding-left: var(--spacing-md); padding-right: var(--spacing-md); box-sizing: border-box; } +header.site .bar { display: flex; flex-wrap: wrap; align-items: center; gap: 0 var(--spacing-lg); } +header.site .brand { font-weight: var(--font-weight-bold); font-size: var(--font-size-lg); color: var(--color-fg-primary); + padding: var(--spacing-sm) 0; letter-spacing: -0.01em; } +header.site .brand:hover { text-decoration: none; color: var(--color-accent); } +header.site .brand::before { content: ""; display: inline-block; width: 1.1em; height: 1.1em; margin-right: 0.3em; vertical-align: -0.1em; + background: url("/favicon.ico") center / contain no-repeat; } +header.site .tagline { color: var(--color-fg-muted); } +.site-nav { display: flex; flex-wrap: wrap; gap: 0 var(--spacing-md); } +.site-nav a { color: var(--color-fg-muted); padding: var(--spacing-sm) 0 calc(var(--spacing-sm) - 2px); border-bottom: 2px solid transparent; } +.site-nav a:hover { color: var(--color-fg-primary); text-decoration: none; } +.site-nav a[aria-current] { color: var(--color-fg-primary); font-weight: var(--font-weight-semibold); border-bottom-color: var(--color-accent); } + +main { padding-top: var(--spacing-lg); padding-bottom: var(--spacing-xl); } +main > * { max-width: 46rem; } +footer.site { margin-top: var(--spacing-xl); border-top: 1px solid var(--color-border-primary); background: var(--color-bg-secondary); + color: var(--color-fg-muted); font-size: var(--font-size-sm); } +footer.site .cols { padding-top: var(--spacing-md); padding-bottom: var(--spacing-lg); } +footer.site p { margin: var(--spacing-xs) 0; } + +/* Plain lists: room for the numbers and bullets, which sit outside the list box (the reset in _shared.css + removes all list padding). Same fix as on code.fuzzyworld.net. */ +main ul:not([class]), main ol:not([class]) { padding-left: 1.75em; margin: 0 0 var(--spacing-md); } +main ul:not([class]) li + li, main ol:not([class]) li + li { margin-top: 0.25em; } + +main h1 { font-size: var(--font-size-2xl); line-height: var(--line-height-tight); margin: 0 0 var(--spacing-sm); } +main h2 { margin-top: var(--spacing-2xl); } +main h3 { margin-top: var(--spacing-xl); color: var(--color-fg-secondary); font-size: var(--font-size-lg); } +.meta { color: var(--color-fg-secondary); font-size: var(--font-size-sm); margin: 0 0 var(--spacing-lg); } +.tag { display: inline-block; padding: 0 var(--spacing-sm); border: 1px solid var(--color-border-primary); border-radius: 1em; text-decoration: none; } +.years { display: flex; flex-wrap: wrap; gap: var(--spacing-md); } + +.post-body img, .post-body video { max-width: 100%; height: auto; } +.post-body blockquote { margin: var(--spacing-md) 0; padding-left: var(--spacing-md); border-left: 3px solid var(--color-border-secondary); color: var(--color-fg-secondary); } +.post-body table { border-collapse: collapse; } +.post-body th, .post-body td { border: 1px solid var(--color-border-primary); padding: var(--spacing-xs) var(--spacing-sm); } +/* From the WordPress "Custom CSS" of the old site: single line breaks in the source do not show (the text runs on, as in + WordPress), a scene break (---) is three centered dots, chapter headings are underlined. The <br> stays in the HTML. */ +.post-body br, .excerpt br { display: none; } +.post-body hr { border: 0; height: auto; margin: var(--spacing-lg) 0; text-align: center; } +.post-body hr::after { content: "\B7\B7\B7"; font-size: 2rem; line-height: 1; letter-spacing: 0.4em; margin-left: 0.4em; color: var(--color-fg-muted); } +.post-body h3, .entry .excerpt h5.chapter { text-decoration: underline; } +pre { overflow-x: auto; padding: var(--spacing-md); background: var(--color-bg-secondary); border-radius: 6px; } + +.entry { margin: var(--spacing-md) 0 var(--spacing-lg); } +.entry h4 { margin: 0; font-size: var(--font-size-lg); } +.entry .meta { margin: 0; } +.entry .excerpt { margin: var(--spacing-xs) 0 0; } +.entry .excerpt p { margin: 0 0 var(--spacing-sm); } +.entry .excerpt h5.chapter { margin: var(--spacing-md) 0 var(--spacing-xs); font-size: var(--font-size-base, 1rem); color: var(--color-fg-secondary); } +.entry .more { margin: 0; } + +.comments-section { margin-top: var(--spacing-2xl); } +ul.comments { list-style: none; padding: 0; } +ul.comments ul.comments { margin-left: var(--spacing-lg); padding-left: var(--spacing-md); border-left: 2px solid var(--color-border-primary); } +.comment { margin: var(--spacing-md) 0; } +.comment p { margin: 0 0 var(--spacing-xs); } + +/* Code colors: the token classes gen.py's highlighter writes. */ +.hl .c { color: var(--color-hl-comment); font-style: italic; } +.hl .cp, .hl .k { color: var(--color-hl-keyword); } +.hl .s { color: var(--color-hl-string); } +.hl .m, .hl .nb, .hl .na { color: var(--color-hl-number); } +.hl .nf, .hl .nd { color: var(--color-hl-function); } +.hl .nc { color: var(--color-hl-class); } +.hl .nt { color: var(--color-hl-type); } +.hl .gi { color: var(--color-success); } +.hl .gd { color: var(--color-error); } +.hl .gh { font-weight: var(--font-weight-bold); } + +/* Light/dark switch (button added by blog.js) and reading progress bar. The color tokens for a chosen theme are + appended to this file at build time from _shared.css (blog.py: theme_overrides). */ +.theme-toggle { margin-left: auto; padding: var(--spacing-xs) var(--spacing-sm); font-size: var(--font-size-lg); line-height: 1; cursor: pointer; + background: none; border: 1px solid var(--color-border-primary); border-radius: 6px; color: var(--color-fg-muted); } +.theme-toggle:hover { color: var(--color-fg-primary); border-color: var(--color-border-secondary); } +.progress { position: fixed; top: 0; left: 0; height: 3px; width: 0; background: var(--color-accent); z-index: 10; pointer-events: none; } diff --git a/server/site-blog/blog.js b/server/site-blog/blog.js new file mode 100644 index 0000000..406f93b --- /dev/null +++ b/server/site-blog/blog.js @@ -0,0 +1,57 @@ +/* The blog's two small conveniences: a light/dark switch and a reading-progress bar. No dependencies. + The choice is kept in localStorage; with none, the system setting decides (see _shared.css). + Loaded in <head> so the saved theme is applied before the first paint. */ +(function () { + var root = document.documentElement, KEY = "fw-theme", saved = null; + try { saved = localStorage.getItem(KEY); } catch (e) {} + if (saved === "light" || saved === "dark") root.setAttribute("data-theme", saved); + + function effective() { + var t = root.getAttribute("data-theme"); + return t || (window.matchMedia && matchMedia("(prefers-color-scheme: dark)").matches ? "dark" : "light"); + } + + document.addEventListener("DOMContentLoaded", function () { + var bar = document.querySelector("header.site .bar"); + if (bar) { + var b = document.createElement("button"); + b.type = "button"; + b.className = "theme-toggle"; + var label = function () { + var dark = effective() === "dark"; + b.textContent = dark ? "☀" : "☾"; + var text = dark ? "Switch to the light theme" : "Switch to the dark theme"; + b.title = text; + b.setAttribute("aria-label", text); + }; + b.addEventListener("click", function () { + var next = effective() === "dark" ? "light" : "dark"; + root.setAttribute("data-theme", next); + try { localStorage.setItem(KEY, next); } catch (e) {} + label(); + }); + label(); + bar.appendChild(b); + if (window.matchMedia) { // the system setting changed: the label follows unless the reader chose a theme + var mq = matchMedia("(prefers-color-scheme: dark)"); + if (mq.addEventListener) mq.addEventListener("change", label); else if (mq.addListener) mq.addListener(label); + } + } + + var post = document.querySelector(".post-body"); + if (post) { + var p = document.createElement("div"); + p.className = "progress"; + p.setAttribute("aria-hidden", "true"); + document.body.appendChild(p); + var update = function () { // 0 at the top of the page, 1 at the bottom + var room = document.documentElement.scrollHeight - window.innerHeight; + var done = room > 0 ? window.scrollY / room : 0; + p.style.width = Math.max(0, Math.min(1, done)) * 100 + "%"; + }; + window.addEventListener("scroll", update, { passive: true }); + window.addEventListener("resize", update); + update(); + } + }); +})(); diff --git a/server/site-blog/htaccess b/server/site-blog/htaccess new file mode 100644 index 0000000..7d477b8 --- /dev/null +++ b/server/site-blog/htaccess @@ -0,0 +1,31 @@ +# writing.fuzzyworld.net. blog.py writes this file as .htaccess, followed by the redirects it generates. +Options -Indexes -Includes -ExecCGI +DirectoryIndex index.html index.xml +ErrorDocument 404 /404.html + +RewriteEngine On +# www.<site> -> <site>, keeping path and query. +RewriteCond %{HTTP_HOST} ^www\.(.+)$ [NC] +RewriteRule ^ https://%1%{REQUEST_URI} [R=301,L] + +# Nothing in this tree is ever executed: pages are pre-generated HTML and media are images and the like. +RemoveHandler .php .php3 .php4 .php5 .php7 .php8 .phtml .phar .shtml .cgi .pl .py +RemoveType .php .php3 .php4 .php5 .php7 .php8 .phtml .phar +<FilesMatch "\.(php\d?|phtml|phar|shtml|cgi|pl|py|sh)$"> + Require all denied +</FilesMatch> + +# The only index.xml is the Atom feed (/feed/index.xml, served at /feed/). +<FilesMatch "^index\.xml$"> + ForceType application/atom+xml +</FilesMatch> + +<IfModule mod_headers.c> + Header always set Content-Security-Policy "default-src 'self'; img-src 'self' https: data:; style-src 'self' 'unsafe-inline'; script-src 'self'; frame-src 'none'; frame-ancestors 'none'" + Header always set X-Content-Type-Options nosniff + Header always set Referrer-Policy strict-origin-when-cross-origin + # Pages change when the blog is rebuilt: browsers revalidate (cheap 304 when unchanged). + <FilesMatch "\.(html|css|js|xml)$"> + Header set Cache-Control "no-cache" + </FilesMatch> +</IfModule> diff --git a/server/site-blog/robots.txt b/server/site-blog/robots.txt new file mode 100644 index 0000000..0374ef2 --- /dev/null +++ b/server/site-blog/robots.txt @@ -0,0 +1,18 @@ +# AI training and AI search crawlers are not welcome here. +User-agent: GPTBot +Disallow: / + +User-agent: ClaudeBot +Disallow: / + +User-agent: CCBot +Disallow: / + +User-agent: Google-Extended +Disallow: / + +User-agent: PerplexityBot +Disallow: / + +User-agent: * +Allow: / diff --git a/tests/test_blog.py b/tests/test_blog.py new file mode 100644 index 0000000..3abd97c --- /dev/null +++ b/tests/test_blog.py @@ -0,0 +1,736 @@ +"""Tests for server/blog.py and server/atom.py. Run: python3 -m unittest discover tests + +Every source file here is made up, in a temporary folder. +""" +import datetime +import json +import os +import sys +import tempfile +import unittest +import xml.etree.ElementTree as ET + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "server")) +import atom # noqa: E402 +import blog # noqa: E402 + +needs_markdown = unittest.skipUnless(blog.markdown, "Python-Markdown is not installed (see FW_PYLIB)") + +POST = """# First Post + +Created by template, on 2018-03-02 + +## Writing + +Line one +line two + +Second paragraph with [a link](https://old.example.org/2018/03/02/first-post/). + +## Spelling + +SECRET-SPELLING-WORDS + +## Removed writing + +SECRET-REMOVED-TEXT +""" + + +class Tree: + """A made-up blog folder: write files, then load_config().""" + + def __init__(self, test, config=None): + self.dir = tempfile.mkdtemp() + test.addCleanup(self.cleanup) + self.src = os.path.join(self.dir, "src") + os.makedirs(self.src) + self.config = {"title": "Test Blog", "base_url": "https://blog.example.org", "source": "src", + "author": "A. Writer", "copyright": "© {years} A. Writer. All rights reserved.", + "rewrite_hosts": ["https://old.example.org"], "nav": [["Blog", "/blog/"]]} + self.config.update(config or {}) + + def cleanup(self): + import shutil + shutil.rmtree(self.dir, ignore_errors=True) + + def file(self, rel, text, base=None): + path = os.path.join(base or self.src, rel) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + f.write(text) + + def cfg(self): + path = os.path.join(self.dir, "fw-blog.json") + with open(path, "w") as f: + json.dump(self.config, f) + return blog.load_config(path) + + def build(self, htaccess=True): + cfg = self.cfg() + out = os.path.join(self.dir, "out") + blog.build(cfg, out, htaccess=htaccess) + return cfg, out + + def read(self, out, rel): + with open(os.path.join(out, rel), encoding="utf-8") as f: + return f.read() + + +def all_output(out): + texts = [] + for dirpath, _, files in os.walk(out): + for name in files: + with open(os.path.join(dirpath, name), "rb") as f: + texts.append(f.read().decode("utf-8", errors="replace")) + return "\n".join(texts) + + +class FrontMatterAndConfig(unittest.TestCase): + def test_front_matter_parsing(self): + meta, rest = blog.parse_front_matter("---\npublished: true\ntitle: \"Hi: there\"\ntags: a, b\n---\n# T\nbody") + self.assertEqual(meta, {"published": True, "title": "Hi: there", "tags": "a, b"}) + self.assertEqual(rest, "# T\nbody") + + def test_horizontal_rule_is_not_front_matter(self): + text = "---\nnot a key value line\n---\ntext" + self.assertEqual(blog.parse_front_matter(text), ({}, text)) + self.assertEqual(blog.parse_front_matter("---\nnever closed"), ({}, "---\nnever closed")) + + def test_unknown_and_missing_config_keys(self): + t = Tree(self, {"bogus": 1}) + with self.assertRaises(blog.BlogError): + t.cfg() + t = Tree(self) + del t.config["title"] + with self.assertRaises(blog.BlogError): + t.cfg() + + def test_timezone_key_is_validated(self): + self.assertEqual(Tree(self).cfg()["timezone"], "UTC") + self.assertEqual(Tree(self, {"timezone": "America/Los_Angeles"}).cfg()["timezone"], "America/Los_Angeles") + for bad in ("Mars/Olympus", "", 5): + with self.assertRaises(blog.BlogError): + Tree(self, {"timezone": bad}).cfg() + + def test_precedence_front_matter_over_config_over_file(self): + t = Tree(self, {"posts": {"2018-03-02-first-post.md": {"title": "Config Title", "slug": "config-slug", "published": True, + "tags": ["x"], "date": "2019-01-01"}}}) + t.file("2018-03-02-first-post.md", "---\ntitle: Front Title\n---\n" + POST) + post = blog.scan_posts(t.cfg())[0] + self.assertEqual(post.title, "Front Title") # front matter beats config + self.assertEqual(post.slug, "config-slug") # config beats file name + self.assertEqual(post.date.date(), datetime.date(2019, 1, 1)) + self.assertTrue(post.published) + self.assertEqual(post.path, "/2019/01/01/config-slug/") + + def test_file_defaults(self): + t = Tree(self) + t.file("2018-03-02-first-post.md", POST) + post = blog.scan_posts(t.cfg())[0] + self.assertEqual((post.title, post.slug, post.path), ("First Post", "first-post", "/2018/03/02/first-post/")) + self.assertFalse(post.published) # nothing says published + + def test_published_only_when_exactly_true(self): + t = Tree(self, {"posts": {"2018-01-01-b.md": {"published": "true"}, "2018-01-02-c.md": {"published": 1}}}) + t.file("2018-01-01-a.md", "---\npublished: yes\n---\n# A\ntext") + t.file("2018-01-01-b.md", "# B\ntext") + t.file("2018-01-02-c.md", "# C\ntext") + t.file("2018-01-03-d.md", "---\npublished: true\n---\n# D\ntext") + t.file("2018-01-04-e.md", "---\npublished: false\n---\n# E\ntext") + got = {p.rel: p.published for p in blog.scan_posts(t.cfg())} + self.assertEqual(got, {"2018-01-01-a.md": False, "2018-01-01-b.md": False, "2018-01-02-c.md": False, + "2018-01-03-d.md": True, "2018-01-04-e.md": False}) + + def test_conflicted_copies_skipped(self): + t = Tree(self) + t.file("2018-03-02-a.md", "# A") + t.file("2018-03-02-a (Pavel's conflicted copy 2018-03-03).md", "# A2") + self.assertEqual([p.rel for p in blog.scan_posts(t.cfg())], ["2018-03-02-a.md"]) + + +class ExtractBody(unittest.TestCase): + def test_writing_section(self): + body = blog.extract_body(POST) + self.assertTrue(body.startswith("Line one")) + for secret in ("SECRET-SPELLING-WORDS", "SECRET-REMOVED-TEXT", "## Spelling", "Created by template", "First Post"): + self.assertNotIn(secret, body) + + def test_without_writing_section(self): + body = blog.extract_body("# Title\n\nCreated by template, on 2018-01-01\n\nJust text.\n\n## Spelling\n\nSECRET\n") + self.assertEqual(body, "Just text.") + + def test_writing_not_last_section(self): + text = "# T\n\n## Writing\n\nBody.\n\n## Info\n\nINFO-TEXT\n\n## Spelling\n\nSECRET\n" + self.assertEqual(blog.extract_body(text), "Body.") + + def test_body_until_override(self): + text = "# T\n\n## Writing\n\nBody.\n\n## Actual writing\n\nMore body.\n\n## Spelling\n\nSECRET\n\n## Info\n\nINFO-TEXT\n" + body = blog.extract_body(text, "Writing", "Info") + self.assertIn("Body.", body) + self.assertIn("More body.", body) + self.assertNotIn("INFO-TEXT", body) + self.assertNotIn("SECRET", body) + + def test_private_sections_dropped_without_writing_section_and_with_section_none(self): + text = "# T\n\nIntro.\n\n## Removed writing\n\nSECRET\n\n## Other\n\nKept.\n" + for body in (blog.extract_body(text), blog.extract_body(text, None)): + self.assertNotIn("SECRET", body) + self.assertIn("Kept.", body) + + def test_html_comments_removed(self): + text = ("# T\n\n## Writing\n\nVisible <!-- inline secret --> text.\n\n<!--\nmulti\nline SECRET-COMMENT\n## Spelling\n-->\n\n" + "Before more.\n<!--more-->\nAfter more.\n\n<!-- never closed SECRET-OPEN\n") + body = blog.extract_body(text) + self.assertNotIn("SECRET", body) + self.assertNotIn("inline secret", body) + self.assertIn("Visible text.", body) + self.assertIn("<!--more-->", body) # the excerpt marker survives for the renderer + + def test_heading_inside_code_fence_is_not_a_section(self): + text = "# T\n\n## Writing\n\n```\n## Spelling\ncode\n```\n\nEnd.\n" + self.assertIn("code", blog.extract_body(text)) + + def test_title_of(self): + self.assertEqual(blog.title_of("intro\n# The Title\n\n## Writing\n"), "The Title") + self.assertIsNone(blog.title_of("no title")) + + +@needs_markdown +class Build(unittest.TestCase): + def blog_tree(self, **config): + t = Tree(self, config) + t.file("2018-03-02-first-post.md", "---\npublished: true\ntags: Fiction, Notes\n---\n" + POST) + t.file("2016-05-06-old-post.md", "---\npublished: true\n---\n# Old\n\nSome old text.\n<!--more-->\nrest\n") + t.file("2019-01-01-draft.md", "# Draft\n\nDRAFT-TEXT\n") + return t + + def test_unpublished_not_built_and_private_text_never_output(self): + t = self.blog_tree() + cfg, out = t.build() + everything = all_output(out) + for secret in ("DRAFT-TEXT", "SECRET-SPELLING-WORDS", "SECRET-REMOVED-TEXT", "Created by template"): + self.assertNotIn(secret, everything) + self.assertFalse(os.path.exists(os.path.join(out, "2019"))) + self.assertTrue(os.path.exists(os.path.join(out, "2018/03/02/first-post/index.html"))) + + def test_nl2br(self): + t = self.blog_tree(line_breaks="br") + _, out = t.build() + self.assertIn("Line one<br", t.read(out, "2018/03/02/first-post/index.html")) + t = self.blog_tree(line_breaks="markdown") + _, out = t.build() + self.assertNotIn("Line one<br", t.read(out, "2018/03/02/first-post/index.html")) + + def test_absolute_links_rewritten_and_redirected(self): + t = Tree(self, {"rewrite_hosts": ["https://old.example.org"], "redirects": {"/2018/03/02/moved/": "/2018/03/02/new/"}, + "wp_ids": {"12": "/about/"}}) + t.file("2018-03-02-p.md", "---\npublished: true\n---\n# P\n\n[a](https://old.example.org/x/y/#frag) " + "[b](https://old.example.org/2018/03/02/moved/) [c](https://elsewhere.org/z/) " + "[d](https://old.example.org/?p=12) \n") + _, out = t.build() + page = t.read(out, "2018/03/02/p/index.html") + self.assertIn('href="/x/y/#frag"', page) + self.assertIn('href="/2018/03/02/new/"', page) + self.assertIn('href="https://elsewhere.org/z/"', page) + self.assertIn('href="/about/"', page) + self.assertIn('src="/wp-content/uploads/a.png"', page) + self.assertNotIn("old.example.org", page) + + def test_atom_feed_parses_and_has_absolute_links(self): + t = self.blog_tree() + _, out = t.build() + root = ET.fromstring(t.read(out, "feed/index.xml").encode("utf-8")) + ns = {"a": atom.NS} + entries = root.findall("a:entry", ns) + self.assertEqual(len(entries), 2) + self.assertEqual(entries[0].find("a:title", ns).text, "First Post") # newest first + self.assertIn("2018", root.find("a:rights", ns).text) + content = entries[0].find("a:content", ns).text + self.assertIn('href="https://blog.example.org/2018/03/02/first-post/"', content) # old host -> site -> absolute + self.assertEqual(blog.absolutize('<a href="/x/">a</a><img src="//cdn/y.png"><a href="https://z/">z</a>', "https://b.org"), + '<a href="https://b.org/x/">a</a><img src="//cdn/y.png"><a href="https://z/">z</a>') + + def test_feed_times_use_the_configured_timezone(self): + def times(**config): + t = Tree(self, config) + t.file("2018-03-02-p.md", "---\npublished: true\ndate: 2018-03-02 09:15:00\n---\n# P\n\ntext\n") + _, out = t.build() + root = ET.fromstring(t.read(out, "feed/index.xml").encode("utf-8")) + entry = root.find("a:entry", {"a": atom.NS}) + return (root.find("a:updated", {"a": atom.NS}).text, entry.find("a:published", {"a": atom.NS}).text) + self.assertEqual(times(), ("2018-03-02T09:15:00Z",) * 2) # default UTC + self.assertEqual(times(timezone="America/Los_Angeles"), ("2018-03-02T17:15:00Z",) * 2) # PST in March + self.assertEqual(times(timezone="Europe/Berlin"), ("2018-03-02T08:15:00Z",) * 2) + self.assertEqual(blog.localize({"timezone": "America/Los_Angeles"}, datetime.datetime(2018, 7, 1, 12)).utcoffset(), + datetime.timedelta(hours=-7)) # PDT in July + + def test_feed_escapes_markup_and_illegal_characters(self): + xml = atom.atom_feed("T & <t>", "https://x.org/", datetime.datetime(2020, 1, 2), + [{"title": "A <b>&", "url": "https://x.org/a/", "updated": datetime.date(2020, 1, 2), + "content": "<p>x\x0c</p>"}], rights="© me") + root = ET.fromstring(xml.encode("utf-8")) + self.assertEqual(root.find("{%s}title" % atom.NS).text, "T & <t>") + self.assertEqual(root.find("{%s}entry/{%s}content" % (atom.NS, atom.NS)).text, "<p>x</p>") + + def test_comments_escaped_threaded_and_minimal(self): + t = self.blog_tree(comments="comments.json") + t.file("comments.json", json.dumps({"/2018/03/02/first-post/": [ + {"id": 1, "parent": 0, "author": "Eve <script>", "date": "2018-03-03T10:00:00", "text": "Hi <b>there</b>\nnew line", + "email": "eve@example.org", "author_url": "https://evil.example"}, + {"id": 2, "parent": 1, "author": "Bob", "date": "2018-03-04T10:00:00", "text": "reply"}]}), base=t.dir) + _, out = t.build() + page = t.read(out, "2018/03/02/first-post/index.html") + self.assertIn("Eve <script>", page) + self.assertIn("Hi <b>there</b><br>", page) + self.assertNotIn("<script>", page) + self.assertNotIn("eve@example.org", page) + self.assertNotIn("evil.example", page) + self.assertIn("Comments © their authors.", page) + self.assertLess(page.index("Eve"), page.index("Bob")) + self.assertIn('<ul class="comments"><li class="comment" id="comment-2"', page) # the reply is nested + + def test_copyright_years(self): + t = self.blog_tree() + _, out = t.build() + self.assertIn("© 2016–2018 A. Writer. All rights reserved.", t.read(out, "blog/index.html")) + self.assertIn("© 2016–2018", t.read(out, "feed/index.xml")) + t = Tree(self) + t.file("2018-03-02-only.md", "---\npublished: true\n---\n# Only\ntext") + _, out = t.build() + self.assertIn("© 2018 A. Writer", t.read(out, "index.html")) + + def test_excerpt_keeps_paragraphs_and_links_to_the_rest(self): + long_para = " ".join(f"w{i}" for i in range(78)) + t = self.blog_tree() + t.file("2018-04-01-long.md", f"---\npublished: true\n---\n# Long One\n\n## Writing\n\n{long_para}\n\nsecond paragraph here\n\nthird one\n") + t.file("2018-04-02-short.md", "---\npublished: true\n---\n# Short One\n\n## Writing\n\nonly a line\n") + _, out = t.build() + index = t.read(out, "blog/index.html") + long_entry = index.split("Long One")[1].split("</article>")[0] + self.assertIn("<p>w0 w1", long_entry) # paragraphs stay paragraphs + self.assertNotIn("second paragraph here", long_entry) # whole paragraphs, up to about 80 words + self.assertIn("Continue reading", long_entry) + self.assertIn('href="/2018/04/01/long/"', long_entry) + short_entry = index.split("Short One")[1].split("</article>")[0] + self.assertIn("<p>only a line</p>", short_entry) + self.assertNotIn("Continue reading", short_entry) # nothing more to read + + def test_excerpt_keeps_chapter_headings(self): + t = self.blog_tree() + t.file("2018-04-03-chapters.md", "---\npublished: true\n---\n# Chapters\n\n## Writing\n\n### Lights & time\n\nfirst words\n\n" + + " ".join(f"w{i}" for i in range(90)) + "\n\n### Second chapter\n\nlater\n") + _, out = t.build() + entry = t.read(out, "blog/index.html").split("Chapters")[1].split("</article>")[0] + self.assertIn('<h5 class="chapter">Lights & time</h5>', entry) # the chapter title shows in the list + self.assertLess(entry.index("Lights"), entry.index("first words")) + self.assertNotIn("Second chapter", entry) # a heading is not left dangling at the end + + def test_theme_switch_and_progress_bar(self): + t = self.blog_tree() + _, out = t.build() + self.assertIn('<script src="/blog.js"></script>', t.read(out, "2018/03/02/first-post/index.html")) + self.assertTrue(os.path.exists(os.path.join(out, "blog.js"))) + css = t.read(out, "blog.css") + shared = t.read(out, "_shared.css") + self.assertIn(".post-body br", css) # as on the old site: line breaks in the source do not show + self.assertIn(':root[data-theme="dark"]', css) + self.assertIn(':root[data-theme="light"]', css) + dark = css.split(':root[data-theme="dark"]')[1] + light = css.split(':root[data-theme="light"]')[1].split(":root[data-theme=")[0] + self.assertIn("--color-bg-primary: #111113;", dark) # the values come from _shared.css + self.assertIn("--color-bg-primary: #ffffff;", light) + self.assertIn("--color-bg-primary: #111113;", shared) + blog.audit_output(out) # .js is an allowed generated file: no BlogError + + def test_index_pages_tags_and_home(self): + t = self.blog_tree(pages_dir="pages", home="start", tags={"fiction": "Fiction"}, feed={"path": "/feed/", "entries": 1}) + t.file("pages/start.md", "---\npublished: true\ntitle: Start Here\n---\nWelcome.\n") + t.file("pages/about.md", "---\npublished: true\n---\n# About\n\nAbout text.\n") + t.file("pages/forgotten.md", "# Forgotten\n\nPAGE-NOT-PUBLISHED\n") # pages are opt-in, like posts + t.file("pages/hidden.md", "---\npublished: false\n---\n# Hidden\n\nPAGE-FALSE\n") + _, out = t.build() + self.assertNotIn("PAGE-NOT-PUBLISHED", all_output(out)) + self.assertNotIn("PAGE-FALSE", all_output(out)) + self.assertFalse(os.path.exists(os.path.join(out, "forgotten"))) + blog_index = t.read(out, "blog/index.html") + self.assertIn('id="2018"', blog_index) + self.assertIn('id="2018-03"', blog_index) + self.assertIn('id="2016-05"', blog_index) + self.assertLess(blog_index.index('id="2018"'), blog_index.index('id="2016"')) + self.assertIn("Some old text.", blog_index) + self.assertNotIn("rest", blog_index.split("Some old text.")[1].split("</article>")[0]) # excerpt stops at <!--more--> + self.assertIn('href="/tag/fiction/"', t.read(out, "2018/03/02/first-post/index.html")) + self.assertIn("Fiction", t.read(out, "tag/fiction/index.html")) + self.assertIn("Welcome.", t.read(out, "index.html")) + self.assertIn("About text.", t.read(out, "about/index.html")) + self.assertTrue(os.path.exists(os.path.join(out, "404.html"))) + root = ET.fromstring(t.read(out, "feed/index.xml").encode("utf-8")) + self.assertEqual(len(root.findall("{%s}entry" % atom.NS)), 1) # feed.entries + + def test_highlighting_fenced_code(self): + t = Tree(self) + t.file("2018-03-02-code.md", "---\npublished: true\n---\n# C\n\n```python\ndef f(): return 1 < 2\n```\n") + _, out = t.build() + page = t.read(out, "2018/03/02/code/index.html") + self.assertIn("<", page) + if blog.gen.pygments_lexers: + self.assertIn('<span class="k">def</span>', page) + + def test_htaccess_and_redirects(self): + t = self.blog_tree(redirects={"/old-page/": "/about/"}, wp_ids={"7": "/2018/03/02/first-post/"}, home="start", + pages_dir="pages") + t.file("pages/start.md", "---\npublished: true\n---\n# Start\ntext") + _, out = t.build() + ht = t.read(out, ".htaccess") + self.assertIn("script-src 'self'", ht) # blog.js only: no inline scripts, no other origins + self.assertIn("frame-src 'none'", ht) + self.assertIn("DirectoryIndex index.html index.xml", ht) + self.assertIn("ForceType application/atom+xml", ht) + self.assertIn("ErrorDocument 404 /404.html", ht) + self.assertIn("RewriteRule ^old-page/?$ /about/ [R=301,L]", ht) + self.assertIn("RewriteCond %{QUERY_STRING} (^|&)p=7(&|$)", ht) + self.assertIn("RewriteCond %{QUERY_STRING} (^|&)page_id=7(&|$)", ht) + self.assertIn("RewriteRule ^start/?$ / [R=301,L]", ht) + self.assertIn("/blog/#$1 [NE,R=301,L]", ht) + self.assertIn("/blog/#$1-$2 [NE,R=301,L]", ht) + self.assertIn("RewriteRule ^(wp-admin|wp-login\\.php|xmlrpc\\.php) - [R=404,L]", ht) + _, out2 = Tree.build(self.blog_tree(), htaccess=False) + self.assertFalse(os.path.exists(os.path.join(out2, ".htaccess"))) + + def test_pattern_escaping(self): + self.assertEqual(blog.apache_pattern("/a.b/c+d/"), r"^a\.b/c\+d/?$") + + +@needs_markdown +class UploadSafety(unittest.TestCase): + """~/Blog holds much that must never be published: only generated files and allowlisted media may be uploaded.""" + + def make(self): + t = Tree(self, {"media": "media", "pages_dir": "pages", "comments": "comments.json"}) + t.file("2018-03-02-first-post.md", "---\npublished: true\n---\n" + POST) + t.file("2018-03-03-draft.md", "# Draft\n\nDRAFT-TEXT\n") + t.file("notes.md", "# Notes\n\nNOTES-TEXT\n") + t.file("pages/about.md", "---\npublished: true\n---\n# About\n\nAbout.\n") + t.file("media/wp-content/uploads/pic.png", "png") + t.file("media/wp-content/uploads/book.epub", "epub") + t.file("media/wp-content/uploads/readme.md", "MEDIA-MD-SECRET") + t.file("media/wp-content/uploads/data.json", "MEDIA-JSON-SECRET") + t.file("media/wp-content/uploads/notes.txt", "MEDIA-TXT-SECRET") + t.file("media/wp-content/uploads/.env", "MEDIA-ENV-SECRET") + t.file("media/wp-content/uploads/run.php", "MEDIA-PHP-SECRET") + t.file("media/_api/posts.json", "MEDIA-API-SECRET") + t.file("wordpress/_api/posts.json", "API-SECRET") + t.file("comments.json", '{"/x/": []}', base=t.dir) + return t + + def test_build_contains_only_allowed_files(self): + t = self.make() + _, out = t.build() + self.assertTrue(os.path.exists(os.path.join(out, "wp-content/uploads/pic.png"))) + self.assertTrue(os.path.exists(os.path.join(out, "wp-content/uploads/book.epub"))) + files = [os.path.relpath(os.path.join(d, f), out) for d, _, fs in os.walk(out) for f in fs] + for f in files: + name = os.path.basename(f) + if name != "robots.txt": + self.assertNotIn(os.path.splitext(name)[1], (".md", ".json", ".txt", ".php", ".env"), f) + everything = all_output(out) + for secret in ("DRAFT-TEXT", "NOTES-TEXT", "SECRET", "fw-blog.json", "comments.json"): + self.assertNotIn(secret, everything) + count, size = blog.audit_output(out) # the audit accepts what the build wrote + self.assertGreater(count, 5) + self.assertGreater(size, 0) + + def test_audit_refuses_stray_files(self): + t = self.make() + _, out = t.build() + for name in ("leak.md", "config.json", "notes.txt", ".env", ".git", "x.php", "noext"): + path = os.path.join(out, name) + with open(path, "w") as f: + f.write("x") + with self.assertRaises(blog.BlogError, msg=name): + blog.audit_output(out) + os.remove(path) + sub = os.path.join(out, "wp-content", "extra.md") + with open(sub, "w") as f: + f.write("x") + with self.assertRaises(blog.BlogError): + blog.audit_output(out) + os.remove(sub) + os.symlink(os.path.join(t.src, "notes.md"), os.path.join(out, "link.png")) + with self.assertRaises(blog.BlogError): + blog.audit_output(out) + + def test_out_inside_source_refused_and_foreign_folder_not_emptied(self): + t = self.make() + cfg = t.cfg() + with self.assertRaises(blog.BlogError): + blog.build(cfg, os.path.join(t.src, "out")) + foreign = os.path.join(t.dir, "precious") + os.makedirs(foreign) + with open(os.path.join(foreign, "keep.txt"), "w") as f: + f.write("x") + with self.assertRaises(blog.BlogError): + blog.build(cfg, foreign) + self.assertTrue(os.path.exists(os.path.join(foreign, "keep.txt"))) + + +SECRET_TAIL = "\n\n## Spelling\n\nSECRET-WORDS\n" + + +class PrivateSectionsFailClosed(unittest.TestCase): + """A private heading written in an unexpected way must never publish its section: it is removed or the build stops.""" + + def body(self, text, **kw): + return blog.extract_body("# T\n\n## Writing\n\n" + text, **kw) + + def test_heading_variants_that_are_not_recognized_stop_the_build(self): + for text in ("Visible.\n\n##Spelling\n\nSECRET-WORDS\n", # no space after ## + "Visible.\n\nSpelling\n--------\n\nSECRET-WORDS\n", # underlined (setext) + "Visible.\n\nRemoved writing\n===\n\nSECRET-WORDS\n", + "Visible.\n\n### Spelling\n\nSECRET-WORDS\n", # wrong level + "Visible.\n\n#Spelling\nSECRET-WORDS\n", + "Visible.\n\n```\nunclosed fence\n\n## Spelling\n\nSECRET-WORDS\n"): + with self.assertRaisesRegex(blog.BlogError, r"my-post\.md"): + self.body(text, name="my-post.md") + + def test_other_section_names_are_caught_when_there_is_no_writing_section(self): + for heading in ("## Spelling:", "## **Spelling**", "## spelling", "## Removed Writing ##"): + text = f"# T\n\nVisible.\n\n{heading}\n\nSECRET-WORDS\n" + try: + body = blog.extract_body(text, None, name="x.md") + except blog.BlogError: + continue + self.assertNotIn("SECRET-WORDS", body, heading) + + def test_error_names_the_line(self): + with self.assertRaisesRegex(blog.BlogError, r"my-post\.md: line 3 .*Spelling"): + self.body("Visible.\n\n##Spelling\nSECRET-WORDS\n", name="my-post.md") + + def test_variants_that_are_removed_cleanly(self): + for text in ("Visible.\n\n ## Spelling\n\nSECRET-WORDS\n", # up to 3 spaces is still a heading + "Visible.\n\n```y``` an inline span\n\n## Spelling\n\nSECRET-WORDS\n", # not a fence + "Visible.\n\n~~~\n```\n~~~\n\n## Removed writing\n\nSECRET-WORDS\n"): + body = self.body(text) + self.assertNotIn("SECRET-WORDS", body) + self.assertIn("Visible.", body) + + def test_fences_follow_commonmark(self): + text = "````\n```\n## Spelling\ncode\n````\n\nAfter.\n\n## Spelling\n\nSECRET-WORDS\n" + body = self.body(text) # the inner ``` does not close the outer fence + self.assertIn("code", body) + self.assertIn("## Spelling\ncode", body) # still code + self.assertIn("After.", body) + self.assertNotIn("SECRET-WORDS", body) + body = self.body("```\n~~~\n## Spelling\ncode\n```\n\nEnd.\n") # another character does not close it either + self.assertIn("code", body) + body = self.body("~~~~\n~~~\n## Spelling\ncode\n~~~~\n\nEnd." + SECRET_TAIL) # a shorter fence does not close it + self.assertIn("code", body) + self.assertNotIn("SECRET-WORDS", body) + + def test_inline_triple_backticks_do_not_open_a_fence(self): + fences = blog._Fences() + self.assertFalse(fences.feed("see ```y``` here")) + self.assertFalse(fences.feed("```y``` here")) + self.assertTrue(fences.feed("```python")) + self.assertTrue(fences.feed("```js")) # an info string cannot close a fence + self.assertIsNotNone(fences.mark) + self.assertTrue(fences.feed("```")) + self.assertIsNone(fences.mark) + + def test_comment_containing_the_more_marker(self): + body = self.body("Before.\n\n<!-- note <!--more--> SECRET-WORDS -->\n\nAfter.\n") + self.assertNotIn("SECRET-WORDS", body) + self.assertNotIn("-->", body) + self.assertNotIn("more", body) + self.assertIn("After.", body) + body = self.body("One.\n<!-- c -->\ntwo <!--more--> three <!-- d --> four\n") # a real marker survives + self.assertEqual(body.count("<!--more-->"), 1) + self.assertNotIn("<!-- c", body) + self.assertIn("three four", body) + + def test_unclosed_comment_warns_naming_the_file(self): + import contextlib + import io + err = io.StringIO() + with contextlib.redirect_stderr(err): + body = self.body("Visible.\n\n<!-- never closed SECRET-WORDS\n", name="draft-one.md") + self.assertIn("draft-one.md", err.getvalue()) + self.assertIn("warning", err.getvalue()) + self.assertNotIn("SECRET-WORDS", body) + err = io.StringIO() + with contextlib.redirect_stderr(err): + self.body("Visible. <!-- closed --> fine\n", name="draft-one.md") + self.assertEqual(err.getvalue(), "") + + def test_a_build_stops_on_a_leaky_heading(self): + t = Tree(self) + t.file("2018-03-02-leaky.md", "---\npublished: true\n---\n# Leaky\n\n## Writing\n\nText.\n\n##Spelling\n\nSECRET-WORDS\n") + with self.assertRaisesRegex(blog.BlogError, "2018-03-02-leaky.md"): + t.build() + self.assertNotIn("SECRET-WORDS", all_output(os.path.join(t.dir, "out")) if os.path.isdir(os.path.join(t.dir, "out")) else "") + + +class SlugsAndPaths(unittest.TestCase): + def test_check_slug(self): + for good in ("first-post", "caf%c3%a9", "v1.2", "A_b", "7"): + self.assertEqual(blog.check_slug(good, "x"), good) + for bad in ("", ".", "..", "../x", "a/b", ".hidden", "a b", "%2e%2e", "a%2fb", "é", "-x", "a\\b", "a%00b", "a\nb"): + with self.assertRaises(blog.BlogError, msg=repr(bad)): + blog.check_slug(bad, "x") + + def test_bad_slugs_in_posts_pages_tags_feed_and_home(self): + t = Tree(self) + t.file("2018-03-02-x.md", "---\npublished: true\nslug: ../../evil\n---\n# X\n\nBody.\n") + with self.assertRaises(blog.BlogError): + t.build() + t = Tree(self, {"pages_dir": "pages"}) + t.file("pages/p.md", "---\npublished: true\nslug: ../evil\n---\n# P\n") + with self.assertRaises(blog.BlogError): + t.build() + for config in ({"tags": {"../x": "Bad"}}, {"feed": {"path": "/../x/"}}, {"feed": {"path": "/a b/"}}, {"home": "../x"}, + {"feed": {"path": "no-slash"}}): + with self.assertRaises(blog.BlogError, msg=str(config)): + Tree(self, config).cfg() + + def test_post_slug_may_be_percent_encoded_and_is_served_decoded(self): + t = Tree(self) + t.file("2018-03-02-x.md", "---\npublished: true\nslug: caf%c3%a9\n---\n# Café\n\nBody.\n") + _, out = t.build() + self.assertTrue(os.path.isfile(os.path.join(out, "2018/03/02/café/index.html"))) # the server decodes %c3%a9 + self.assertIn('href="/2018/03/02/caf%c3%a9/"', t.read(out, "blog/index.html")) + + def test_output_never_leaves_the_out_folder(self): + t = Tree(self, {"url": "/../../escaped/{slug}/"}) + t.file("2018-03-02-x.md", "---\npublished: true\n---\n# X\n\nBody.\n") + with self.assertRaises(blog.BlogError): + t.build() + self.assertFalse(os.path.exists(os.path.join(t.dir, "escaped"))) + out = os.path.join(t.dir, "o") + os.makedirs(os.path.join(t.dir, "elsewhere")) + os.makedirs(out) + os.symlink(os.path.join(t.dir, "elsewhere"), os.path.join(out, "linked")) + with self.assertRaises(blog.BlogError): + blog.safe_join(out, "linked", "index.html") + self.assertEqual(blog.safe_join(out, "a", "b.html"), os.path.join(out, "a", "b.html")) + + +class DeployTarget(unittest.TestCase): + def target(self, deploy): + t = Tree(self, {"deploy": deploy}) + return blog.deploy_target(t.cfg()) + + def test_good_targets(self): + self.assertEqual(self.target({"host": "dh", "dir": "writing.example.org"}), "dh:writing.example.org") + self.assertEqual(self.target({"host": "me@dh", "dir": "sites/writing.example.org/"}), "me@dh:sites/writing.example.org") + + def test_dangerous_dirs_are_refused(self): + for folder in ("", "~", ".", "/", "..", "../x", "a/../b", "a/./b", "-rf", "-x/y", "/abs/path", "~/x", "a b", "x;y", + "a//b", "$HOME"): + with self.assertRaises(blog.BlogError, msg=repr(folder)): + self.target({"host": "dh", "dir": folder}) + for host in ("", "-oProxyCommand=x", "a b"): + with self.assertRaises(blog.BlogError, msg=repr(host)): + self.target({"host": host, "dir": "site.example.org"}) + with self.assertRaises(blog.BlogError): + self.target(None) + + def test_command_exits_non_zero_and_prints_nothing(self): + t = Tree(self, {"deploy": {"host": "dh", "dir": "~"}}) + path = os.path.join(t.dir, "fw-blog.json") + t.cfg() + with self.assertRaises(SystemExit) as caught: + blog.main(["blog.py", "deploy-target", path]) + self.assertNotEqual(caught.exception.code, 0) + self.assertIn("deploy.dir", str(caught.exception.code)) + + +class HtaccessValues(unittest.TestCase): + def test_values_with_spaces_or_control_characters_are_refused(self): + self.assertTrue(blog.htaccess_unsafe("/a b/")) + self.assertTrue(blog.htaccess_unsafe("/a%20b/")) + self.assertTrue(blog.htaccess_unsafe("/a%0Ad/")) + self.assertTrue(blog.htaccess_unsafe("/a\tb")) + self.assertTrue(blog.htaccess_unsafe("/a%0d%0aHeader: x")) + self.assertFalse(blog.htaccess_unsafe("/caf%c3%a9/?x=1")) + for config in ({"redirects": {"/a%0Ab/": "/x/"}}, {"redirects": {"/a/": "/x%20y/"}}, {"redirects": {"/a/": "/x\nRewriteRule"}}, + {"wp_ids": {"7": "/x%0a/"}}, {"wp_ids": {"7%0a": "/x/"}}, {"wp_ids": {"7": "/a b/"}}, + {"feed": {"path": "/f%0a/"}}): + with self.assertRaises(blog.BlogError, msg=str(config)): + Tree(self, config).cfg() + Tree(self, {"redirects": {"/caf%c3%a9/": "/x/"}, "wp_ids": {"7": "/y/"}}).cfg() # fine + + +@needs_markdown +class SmallerFixes(unittest.TestCase): + def test_excerpt_of_a_chapter_heading_and_a_long_paragraph(self): + t = Tree(self) + t.file("2018-04-03-chapter.md", "---\npublished: true\n---\n# Opening\n\n## Writing\n\n### Chapter one\n\n" + + " ".join(f"w{i}" for i in range(120)) + "\n\nsecond paragraph\n") + _, out = t.build() + entry = t.read(out, "blog/index.html").split("Opening")[1].split("</article>")[0] + self.assertIn('<h5 class="chapter">Chapter one</h5>', entry) + self.assertIn("<p>w0 w1", entry) # not an empty excerpt + self.assertIn("w79…", entry) + self.assertNotIn("w80", entry) + self.assertNotIn("second paragraph", entry) + self.assertIn("Continue reading", entry) + + def test_tags_in_brackets(self): + self.assertEqual(blog.as_list('[a, "b c", \'d\']'), ["a", "b c", "d"]) + self.assertEqual(blog.as_list("[]"), []) + self.assertEqual(blog.as_list("a, b"), ["a", "b"]) + self.assertEqual(blog.as_list(["x", " y "]), ["x", "y"]) + t = Tree(self) + t.file("2018-03-02-x.md", "---\npublished: true\ntags: [Fiction, \"Short Stories\"]\n---\n# X\n\nBody.\n") + self.assertEqual(blog.scan_posts(t.cfg())[0].tags, ["Fiction", "Short Stories"]) + + def test_front_matter_after_a_byte_order_mark(self): + t = Tree(self) + t.file("2018-03-02-x.md", "\ufeff---\npublished: true\ntitle: With BOM\n---\n# X\n\nBody.\n") + post = blog.scan_posts(t.cfg())[0] + self.assertTrue(post.published) + self.assertEqual(post.title, "With BOM") + + def test_media_symlinks_are_not_followed(self): + t = Tree(self, {"media": "media"}) + t.file("media/real.png", "png") + outside = os.path.join(t.dir, "outside") + t.file("private.png", "PRIVATE-IMAGE", base=outside) + t.file("secret/other.png", "PRIVATE-DIR-IMAGE", base=outside) + os.symlink(os.path.join(outside, "private.png"), os.path.join(t.src, "media", "link.png")) + os.symlink(os.path.join(outside, "secret"), os.path.join(t.src, "media", "linked-dir")) + t.file("2018-03-02-x.md", "---\npublished: true\n---\n# X\n\nBody.\n") + _, out = t.build() + self.assertTrue(os.path.exists(os.path.join(out, "real.png"))) + self.assertFalse(os.path.lexists(os.path.join(out, "link.png"))) + self.assertFalse(os.path.lexists(os.path.join(out, "linked-dir"))) + self.assertNotIn("PRIVATE", all_output(out)) + + def test_fw_blog_build_passes_options_through_and_rejects_unknown_ones(self): + import subprocess + fw = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "bin", "fw") + env = dict(os.environ, FW_PYLIB=blog.gen.PYLIB, TMPDIR=tempfile.mkdtemp()) + self.addCleanup(__import__("shutil").rmtree, env["TMPDIR"], True) + if subprocess.run(["python3", "-c", "import markdown, pygments"], env=dict(env, PYTHONPATH=blog.gen.PYLIB)).returncode: + self.skipTest("python3 on the PATH cannot import Markdown and Pygments from FW_PYLIB") + t = Tree(self) + t.file("2018-03-02-x.md", "---\npublished: true\n---\n# X\n\nBody.\n") + t.cfg() + config = os.path.join(t.dir, "fw-blog.json") + out = os.path.join(t.dir, "built") + run = lambda *args: subprocess.run(["bash", fw, "blog", "build", config, *args], env=env, capture_output=True, text=True) + r = run("--out", out, "--no-htaccess") + self.assertEqual(r.returncode, 0, r.stderr) + self.assertTrue(os.path.exists(os.path.join(out, "index.html"))) + self.assertFalse(os.path.exists(os.path.join(out, ".htaccess"))) + r = run("--out", out) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertTrue(os.path.exists(os.path.join(out, ".htaccess"))) + r = run("--out", out, "--bogus") + self.assertEqual(r.returncode, 2) + self.assertIn("unknown option --bogus", r.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_wp_import.py b/tests/test_wp_import.py new file mode 100644 index 0000000..24eabf3 --- /dev/null +++ b/tests/test_wp_import.py @@ -0,0 +1,555 @@ +"""Tests for lib/wp_import.py. Run: python3 -m unittest discover tests + +Every post, file and comment here is made up, in a temporary folder. No network: the API data is written +straight into the importer's cache folder, and rsync and HEAD requests are replaced by fakes. +""" +import hashlib +import json +import os +import re +import sys +import tempfile +import unittest +import urllib.error + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.join(HERE, "..", "lib")) +import wp_import as wp # noqa: E402 + +blog = wp.blog +needs_markdown = unittest.skipUnless(blog.markdown, "Python-Markdown is not installed (see FW_PYLIB)") + +WP_URL = "https://old.example.org" +VOCAB = ("river stone lantern copper window harbor violet engine paper meadow signal orchard thunder velvet " + "compass ladder whistle marble garden anchor falcon mirror candle bridge pebble saddle winter ribbon " + "kettle summit shadow timber walnut yonder zephyr button carpet dragon ember feather glacier").split() + + +def words(n, seed=0): + """n made-up words, different for each seed.""" + return " ".join(VOCAB[(i * (seed + 3) + seed * 7) % len(VOCAB)] + ("" if i % 5 else str(seed)) for i in range(n)) + + +def md(html_text): + return wp.html_to_markdown(html_text) + + +class Converter(unittest.TestCase): + def test_wordpress_additions_and_private_comments_removed(self): + text = md('<div id="bsf_rt_marker"></div><span class="span-reading-time rt-reading-time"><span class="rt-label">' + '[Reading time: 3 minutes]</span></span><p><!-- SECRET-CUT [x](y) --></p><p>Visible<span id="more-5"></span> text.</p>' + '<p><!--more--></p><p>After</p><script async src="https://x.example/p.js"></script>') + self.assertNotIn("SECRET-CUT", text) + self.assertNotIn("Reading time", text) + self.assertNotIn("script", text) + self.assertEqual(text, "Visible text.\n\n<!--more-->\n\nAfter") + + def test_line_breaks_and_paragraphs(self): + self.assertEqual(md("<p>one<br />\ntwo<br />\nthree</p>\n<p>next</p>"), "one\ntwo\nthree\n\nnext") + # two <br> in a row: the second stays a literal <br> so the count of <br> is kept + out = md("<p>one<br /><br />two</p>") + self.assertEqual(out.count("\n") + out.count("<br>"), 2) + self.assertEqual(md("<p>x<br /></p>"), "x") + + def test_inline_markup(self): + self.assertEqual(md('<p><em>a</em> <strong>b</strong> <i>c</i> <code>d</code> <del>e</del> ' + '<a href="https://old.example.org/x/y/">link</a></p>'), + "*a* **b** *c* `d` <del>e</del> [link](https://old.example.org/x/y/)") + self.assertEqual(md("<p><em>line one<br />line two</em></p>"), "*line one*\n*line two*") + self.assertEqual(md('<p><a href="https://x.example/a b(1)">t</a> <img src="/u/a.png" alt="An [alt]"></p>'), + "[t](https://x.example/a%20b%281%29) ") + + def test_special_characters_are_escaped(self): + self.assertEqual(md("<p>2 * 3 < 5 & _x_ snake_case</p>"), "2 \\* 3 < 5 & \\_x\\_ snake_case") + self.assertEqual(md("<p># not a heading<br />- not a list<br />1. not numbered<br />> not a quote</p>"), + "\\# not a heading\n\\- not a list\n1\\. not numbered\n\\> not a quote") + self.assertEqual(md("<p>&amp; stays</p>"), "&amp; stays") + + def test_blocks(self): + out = md("<h2>Top</h2><h3>Mid</h3><h4>Low</h4><hr class='x'><blockquote><p>q1</p><p>q2</p></blockquote>" + "<ul><li>a<ul><li>nested</li></ul></li><li>b</li></ul><ol><li>x</li><li>y</li></ol>" + "<pre><code class=\"language-python\">print('hi')\n```\n</code></pre>") + self.assertIn("### Top\n\n### Mid\n\n#### Low\n\n---\n\n> q1\n>\n> q2", out) # no ## heading: it would end the body + self.assertIn("- a\n - nested\n- b", out) + self.assertIn("1. x\n2. y", out) + self.assertIn("````python\nprint('hi')\n```\n````", out) + + def test_embeds_become_links(self): + out = md('<div class="jetpack-video-wrapper"><span class="embed-youtube"><iframe class="youtube-player" ' + 'src="https://www.youtube.com/embed/abc_123?version=3&rel=1"></iframe></span></div>' + '<p><a data-flickr-embed="true" href="https://flickr.example/p/1" title="A photo">' + '<img src="https://flickr.example/i.jpg" alt="A photo"></a><script async src="x.js"></script></p>' + '<div class="tumblr-post" data-href="h"><a href="https://tumblr.example/post/1">https://tumblr.example/post/1</a></div>') + self.assertIn("[Video: https://www.youtube.com/watch?v=abc_123](https://www.youtube.com/watch?v=abc_123)", out) + self.assertIn("[A photo](https://flickr.example/p/1)", out) + self.assertNotIn("flickr.example/i.jpg", out) + self.assertIn("[https://tumblr.example/post/1](https://tumblr.example/post/1)", out) + notes = [] + wp.html_to_markdown('<iframe src="https://other.example/frame"></iframe>', notes) + self.assertEqual(notes, ["iframe dropped: https://other.example/frame"]) + + def test_text_and_br_count_ignore_what_is_removed(self): + h = '<span class="span-reading-time">Reading time 1</span><p>a<br>b <!-- cut --> c</p><p>d</p>' + self.assertEqual(wp.norm_words(wp.html_text(h)), ["a", "b", "c", "d"]) + self.assertEqual(wp.html_br_count(h), 1) + self.assertEqual(wp.html_plain("<p>one<br />two</p><p>three & four</p>"), "one\ntwo\n\nthree & four") + + def test_normalized_words_ignore_typography(self): + self.assertEqual(wp.norm_words("It’s “quoted” – and… done--ok"), wp.norm_words("It's \"quoted\" - and... done - ok")) + self.assertEqual(wp.md_words("Some *text* with [a link](https://x.example/q_r) and  <!--more--> end"), + ["some", "text", "with", "a", "link", "and", "end"]) + + @needs_markdown + def test_round_trip_through_the_generator(self): + h = ("<p>First line<br />\nsecond *line* with 2 * 3 and _under_.</p><h3>Head</h3><p><em>Emphasis</em> and " + "<strong>strong</strong>, <a href=\"https://old.example.org/a/\">a link</a>.</p><ul><li>one</li><li>two</li></ul>" + "<blockquote><p>quoted words</p></blockquote><p>1. not a list</p>") + text = "# T\n\n## Writing\n\n" + md(h) + out = blog.render_markdown(blog.extract_body(text), {"line_breaks": "br", "rewrite_hosts": [], "redirects": {}, + "wp_ids": {}}) + self.assertGreaterEqual(wp.similarity(wp.norm_words(wp.html_text(h)), wp.norm_words(blog.plain_text(out))), 0.99) + self.assertEqual(len(re.findall(r"<br", out)), wp.html_br_count(h)) + self.assertNotIn("<ol", out) + + +def post_json(i, title, slug, date, html_text, tags=(), protected=False, status="publish"): + y, m, d = date[:10].split("-") + return {"id": i, "date": date, "date_gmt": date.replace("T09", "T16"), "slug": slug, "status": status, "type": "post", + "link": f"{WP_URL}/{y}/{m}/{d}/{slug}/", "title": {"rendered": title}, + "content": {"rendered": "" if protected else html_text, "protected": protected}, + "tags": list(tags), "class_list": ["post-password-required"] if protected else []} + + +def comment_json(i, post, author, text, parent=0, status="approved"): + return {"id": i, "post": post, "parent": parent, "author_name": author, "author_url": "https://spam.example/u", + "author_email": "secret@example.org", "author_avatar_urls": {"24": "https://gravatar.example/a"}, + "date": "2018-04-01T10:00:00", "status": status, "type": "comment", "content": {"rendered": f"<p>{text}</p>"}} + + +def _sha(path): + with open(path, "rb") as f: + return hashlib.sha1(f.read()).hexdigest() + + +class Site: + """A made-up blog folder with a made-up WordPress API cache.""" + + def __init__(self, test, config=None): + self.test = test + self.dir = tempfile.mkdtemp() + test.addCleanup(lambda: __import__("shutil").rmtree(self.dir, ignore_errors=True)) + self.src = os.path.join(self.dir, "src") + os.makedirs(self.src) + self.config_path = os.path.join(self.dir, "fw-blog.json") + self.config = {"title": "Test", "base_url": "https://new.example.org", "source": "src", "line_breaks": "br", + "author": "A. Writer"} + self.config.update(config or {}) + self.save_config() + self.api = os.path.join(self.src, "wordpress", "_api") + self.fetch_calls = [] + + def save_config(self): + with open(self.config_path, "w") as f: + json.dump(self.config, f) + + def file(self, rel, text): + path = os.path.join(self.src, rel) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + f.write(text) + + def read(self, rel): + with open(os.path.join(self.src, rel), encoding="utf-8") as f: + return f.read() + + def raw_config(self): + with open(self.config_path) as f: + return json.load(f) + + def api_data(self, posts=(), pages=(), comments=(), tags=(), media=()): + os.makedirs(self.api, exist_ok=True) + for name, data in (("posts", posts), ("pages", pages), ("comments", comments), ("tags", tags), ("media", media)): + with open(os.path.join(self.api, name + ".json"), "w") as f: + json.dump(list(data), f) + + def snapshot(self): + """{relative path: hash} of every file under the folder (config included).""" + out = {} + for root, _, names in os.walk(self.dir): + for n in names: + p = os.path.join(root, n) + out[os.path.relpath(p, self.dir)] = _sha(p) + return out + + def run(self, **kw): + kw.setdefault("no_media", True) + kw.setdefault("say", lambda s: None) + return wp.run_import(self.config_path, WP_URL, **kw) + + +def standard_site(test, **config): + """Two matched-or-not public posts, one protected post, one page, one tag, comments.""" + t = Site(test, config) + first_body = words(60, 1) + second_body = words(60, 2) + t.file("2018-03-02-first-post.md", f"# First Post\n\n## Writing\n\n{first_body}\n\n## Spelling\n\nSECRET-SPELLING\n") + t.file("2018-03-02-first-post (conflicted copy).md", f"# First Post\n\n## Writing\n\n{first_body}\n") + t.file("2017-01-01-unrelated.md", f"# Unrelated\n\n## Writing\n\n{words(60, 9)}\n") + t.api_data( + posts=[post_json(11, "First Post", "first-post", "2018-03-02T09:15:00", f"<p>{first_body}</p>", tags=[5]), + post_json(12, "Second – Post", "second-post", "2018-04-01T09:30:00", + f"<div id=\"bsf_rt_marker\"></div><p>{second_body}<br />\nsecond line</p><p><!--more--></p><p>tail text</p>"), + post_json(13, "Locked", "locked", "2018-05-01T09:00:00", "", protected=True)], + pages=[{"id": 21, "slug": "highlights", "status": "publish", "link": WP_URL + "/", "title": {"rendered": "Highlights"}, + "content": {"rendered": '<p>Front <a href="https://old.example.org/2018/04/01/second-post/">page</a>.</p>', "protected": False}}, + {"id": 22, "slug": "blog", "status": "publish", "link": WP_URL + "/blog/", "title": {"rendered": "Blog"}, + "content": {"rendered": "<p>generated</p>", "protected": False}}], + comments=[comment_json(1, 12, "Pat", "Nice & good<br />really"), comment_json(2, 12, "Sam", "Reply", parent=1), + comment_json(3, 13, "Eve", "On the locked post"), comment_json(4, 12, "Bot", "pending", status="hold")], + tags=[{"id": 5, "slug": "fiction", "name": "Fiction", "link": WP_URL + "/tag/fiction/"}]) + return t + + +class Matching(unittest.TestCase): + def files(self, *specs): + import datetime + return [{"rel": rel, "words": wp.norm_words(text), "title_words": wp.norm_words(title), + "date": datetime.datetime.fromisoformat(date), "post": None} for rel, title, date, text in specs] + + def post(self, title, date, text): + import datetime + return {"title": title, "date": datetime.datetime.fromisoformat(date), "words": wp.norm_words(text), "path": "/" + title} + + def test_threshold_each_file_once_and_double_claims(self): + body = words(80, 1) + files = self.files(("a.md", "Same", "2018-01-01", body), ("far.md", "Other", "2018-01-01", words(80, 4))) + near = body.replace("river", "rivers", 1) # one word differs: well above 0.95 + posts = [self.post("Same", "2018-02-01", body), self.post("Copy", "2018-02-02", near), + self.post("Far", "2018-02-03", words(80, 6))] + assigned, best, conflicts = wp.match_posts(posts, files) + self.assertEqual(sorted(assigned), [0]) # exact match wins, equal title; the copy gets nothing + self.assertEqual(assigned[0][1], 0) + self.assertTrue(any("claimed by" in c and "a.md" in c for c in conflicts)) + self.assertNotIn(2, assigned) # no similar file + files = self.files(("a.md", "Same", "2018-01-01", body)) + half = " ".join(body.split()[:60]) + " " + words(20, 7) # about 75 percent similar + assigned, best, _ = wp.match_posts([self.post("Same", "2018-02-01", half)], files) + self.assertEqual(assigned, {}) + self.assertTrue(0.5 < best[0][0][0] < wp.MATCH_THRESHOLD) # the score is still reported + + def test_ties_go_to_the_equal_title_then_the_earlier_date(self): + body = words(50, 2) + files = self.files(("x.md", "Different title", "2018-01-01", body), ("y.md", "The Title", "2018-01-05", body), + ("z.md", "The Title", "2018-03-01", body)) + assigned, _, conflicts = wp.match_posts([self.post("The Title", "2018-02-01", body)], files) + self.assertEqual(files[assigned[0][1]]["rel"], "y.md") # equal title, dated before publication, nearest + self.assertTrue(any("matches several files" in c for c in conflicts)) + + +class Fetching(unittest.TestCase): + def test_pagination_cache_and_refresh(self): + tmp = tempfile.mkdtemp() + self.addCleanup(lambda: __import__("shutil").rmtree(tmp, ignore_errors=True)) + calls = [] + + def get_json(url): + calls.append(url) + page = int(re.search(r"[?&]page=(\d+)", url).group(1)) + if page > 2: + raise urllib.error.HTTPError(url, 400, "bad page", {}, None) + return [{"id": page * 1000 + i} for i in range(100 if page == 1 else 3)], 2 + + f = wp.Fetcher(WP_URL, os.path.join(tmp, "_api"), get_json=get_json) + self.assertEqual(len(f.collection("posts")), 103) + self.assertEqual(len(calls), 2) + self.assertIn("per_page=100", calls[0]) + wp.Fetcher(WP_URL, os.path.join(tmp, "_api"), get_json=get_json).collection("posts") + self.assertEqual(len(calls), 2) # cache used + wp.Fetcher(WP_URL, os.path.join(tmp, "_api"), refresh=True, get_json=get_json).collection("posts") + self.assertEqual(len(calls), 4) + legacy = os.path.join(tmp, "legacy") + os.makedirs(legacy) + for name in ("p1.json", "p2.json"): # the development cache's names + with open(os.path.join(legacy, name), "w") as fh: + json.dump([{"id": name}], fh) + self.assertEqual(wp.Fetcher(WP_URL, legacy).collection("posts"), [{"id": "p1.json"}, {"id": "p2.json"}]) + + +class Import(unittest.TestCase): + def test_dry_run_writes_only_api_and_report(self): + t = standard_site(self) + before = t.snapshot() + report = t.run(dry_run=True) + after = t.snapshot() + new = sorted(set(after) - set(before)) + self.assertEqual(new, ["src/wordpress/_report.txt"]) + self.assertEqual({k: v for k, v in before.items() if k in after and after[k] != v}, {}) + self.assertIn('"Locked"', report) # skipped posts are listed + self.assertIn("password-protected", report) + self.assertIn("DRY RUN", report) + self.assertRegex(report, r"1\.0000 2018-03-02-first-post\.md <- \"First Post\"") + self.assertNotIn("conflicted copy", report) + self.assertIn("wordpress/2018-04-01-second-post.md", report) + + def test_import_writes_files_comments_and_config(self): + t = standard_site(self) + t.run() + second = t.read("wordpress/2018-04-01-second-post.md") + self.assertEqual(second.split("\n")[0], "# Second – Post") + self.assertIn("## Writing\n\n", second) + self.assertIn("second line", second) + self.assertNotIn("bsf_rt", second) + self.assertIn("<!--more-->", second) + self.assertTrue(t.read("pages/highlights.md").startswith( + "---\npublished: true\n---\n# Highlights\n\nFront [page](https://old.example.org/")) # pages are opt-in + self.assertFalse(os.path.exists(os.path.join(t.src, "pages", "blog.md"))) + self.assertFalse(os.path.exists(os.path.join(t.src, "wordpress", "2018-03-02-first-post.md"))) # matched: the source file is used + self.assertFalse(os.path.exists(os.path.join(t.src, "comments.json"))) # relative to the config, not to `source` + with open(os.path.join(t.dir, "comments.json"), encoding="utf-8") as f: + comments = json.load(f) + self.assertEqual(list(comments), ["/2018/04/01/second-post/"]) + self.assertEqual(comments["/2018/04/01/second-post/"][0], + {"id": 1, "parent": 0, "author": "Pat", "date": "2018-04-01T10:00:00", "text": "Nice & good\nreally"}) + self.assertEqual(sorted(comments["/2018/04/01/second-post/"][1]), ["author", "date", "id", "parent", "text"]) + raw = t.raw_config() + self.assertNotIn("secret", json.dumps(comments)) + self.assertNotIn("spam.example", json.dumps(comments)) + self.assertEqual(raw["posts"]["2018-03-02-first-post.md"], { + "published": True, "date": "2018-03-02T09:15:00", "slug": "first-post", "wp_id": 11, "tags": ["fiction"]}) + self.assertEqual(raw["posts"]["wordpress/2018-04-01-second-post.md"]["wp_id"], 12) + self.assertEqual(raw["tags"], {"fiction": "Fiction"}) + self.assertEqual(raw["wp_ids"], {"11": "/2018/03/02/first-post/", "12": "/2018/04/01/second-post/", "21": "/"}) + self.assertEqual((raw["pages_dir"], raw["home"], raw["comments"], raw["timezone"]), + ("pages", "highlights", "comments.json", "America/Los_Angeles")) + self.assertNotIn("media", raw) # run with no_media + self.assertIn("wordpress/*.md", raw["include"]) + self.assertIn(WP_URL, raw["rewrite_hosts"]) + self.assertNotIn("locked", json.dumps(raw).lower()) # protected: no entries anywhere + self.assertNotIn("On the locked post", json.dumps(comments)) + blog.load_config(t.config_path) # still a valid config + + def test_idempotent_and_never_overwrites_without_force(self): + t = standard_site(self) + t.run() + snap = t.snapshot() + t.run() + self.assertEqual(t.snapshot(), snap) + edited = t.read("wordpress/2018-04-01-second-post.md") + "\nMy own edit.\n" + t.file("wordpress/2018-04-01-second-post.md", edited) + raw = t.raw_config() + raw["posts"]["wordpress/2018-04-01-second-post.md"]["slug"] = "mine" + t.config = raw + t.save_config() + snap = t.snapshot() + report = t.run() + changed = {k for k, v in t.snapshot().items() if snap.get(k) != v} + self.assertEqual(changed, {"src/wordpress/_report.txt"}) # only the report (it now lists the kept file) + self.assertIn("Files kept because they differ", report) + self.assertIn(" wordpress/2018-04-01-second-post.md", report) + self.assertIn("kept existing slug", report) + t.run(force=True) + self.assertNotIn("My own edit.", t.read("wordpress/2018-04-01-second-post.md")) + self.assertEqual(t.raw_config()["posts"]["wordpress/2018-04-01-second-post.md"]["slug"], "second-post") + + def test_existing_config_entries_and_body_until_are_kept(self): + t = standard_site(self, posts={"2018-03-02-first-post.md": {"published": False, "body_until": "Info", "title": "Mine"}}, + tags={"fiction": "My fiction"}, home="elsewhere") + t.run() + raw = t.raw_config() + entry = raw["posts"]["2018-03-02-first-post.md"] + self.assertEqual((entry["published"], entry["body_until"], entry["title"]), (False, "Info", "Mine")) + self.assertEqual(entry["wp_id"], 11) # missing keys are added + self.assertEqual((raw["tags"], raw["home"]), ({"fiction": "My fiction"}, "elsewhere")) + + def test_second_source_file_with_same_text_is_flagged_not_used_twice(self): + t = standard_site(self) + os.remove(os.path.join(t.src, "2018-03-02-first-post (conflicted copy).md")) + t.file("2018-03-03-first-post-again.md", t.read("2018-03-02-first-post.md")) + report = t.run() + self.assertIn("matches several files", report) + self.assertEqual(sum(1 for line in report.split("\n") if re.match(r"\s+1\.0000", line)), 1) + + def test_title_difference_is_recorded(self): + t = standard_site(self) + t.file("2018-03-02-first-post.md", t.read("2018-03-02-first-post.md").replace("# First Post", "# Working Title")) + t.run() + self.assertEqual(t.raw_config()["posts"]["2018-03-02-first-post.md"]["title"], "First Post") + + def test_media_sync_command_is_restricted_and_inside_the_source(self): + t = standard_site(self) + calls = [] + + class Result: + returncode, stderr = 0, "" + + def fake(cmd, **kw): + calls.append(cmd) + return Result() + t.run(no_media=False, rsync=fake, media_from="dh:old.example.org/wp-content/uploads/") + self.assertEqual(len(calls), 1) + cmd = calls[0] + self.assertEqual(cmd[0], "rsync") + self.assertNotIn("--delete", cmd) + self.assertEqual(cmd[-2], "dh:old.example.org/wp-content/uploads/") + self.assertTrue(cmd[-1].endswith("/")) + self.assertEqual(os.path.realpath(cmd[-1]), os.path.realpath(os.path.join(t.src, "media", "wp-content", "uploads"))) + self.assertEqual(t.raw_config()["media"], "media") + includes = [c for c in cmd if c.startswith("--include")] + self.assertIn("--include=*.[jJ][pP][gG]", includes) + self.assertIn("--exclude=*", cmd) + self.assertEqual(cmd[cmd.index("--exclude=*") + 1:cmd.index("--exclude=*") + 2], [cmd[-2]]) + for c in includes: + self.assertNotIn("md", c.lower().replace("*", "")) # no Markdown, JSON or script patterns + t.run(no_media=False, rsync=fake) + self.assertEqual(calls[1][-2], "dh:old.example.org/wp-content/uploads/") # default derived from --wp + n = len(calls) + t.run(dry_run=True, no_media=False, rsync=fake) + self.assertEqual(len(calls), n) # a dry run never copies + + def test_probe_links_only_with_the_flag_and_records_301_targets(self): + t = standard_site(self) + t.api_data(posts=json.load(open(os.path.join(t.api, "posts.json"))) + [ + post_json(14, "Linky", "linky", "2018-06-01T09:00:00", + '<p><a href="https://old.example.org/2017/old-name/">old</a> <a href="/gone/">gone</a> ' + '<a href="https://old.example.org/2018/04/01/second-post/">fine</a> <a href="https://elsewhere.example/x/">out</a> ' + '<img src="https://old.example.org/wp-content/uploads/2018/06/missing.png"></p>')], + pages=json.load(open(os.path.join(t.api, "pages.json"))), + comments=json.load(open(os.path.join(t.api, "comments.json"))), + tags=json.load(open(os.path.join(t.api, "tags.json")))) + heads = [] + + def head(url): + heads.append(url) + return {WP_URL + "/2017/old-name/": (301, WP_URL + "/2018/03/02/first-post/")}.get(url, (404, None)) + + fetcher = wp.Fetcher(WP_URL, t.api, head=head) + report = t.run(dry_run=True, fetcher=fetcher) + self.assertEqual(heads, []) + self.assertIn("unchecked (use --probe-links): /2017/old-name/", report) + self.assertIn("missing media: /wp-content/uploads/2018/06/missing.png", report) + self.assertNotIn("elsewhere.example", report) + report = t.run(dry_run=True, probe=True, fetcher=fetcher) + self.assertEqual(sorted(heads), [WP_URL + "/2017/old-name/", WP_URL + "/gone/"]) + self.assertIn("redirect /2017/old-name/ -> /2018/03/02/first-post/", report) + self.assertIn("/gone/: HTTP 404", report) + self.assertEqual(t.raw_config().get("redirects"), None) # a dry run does not touch the config + t.run(probe=True, fetcher=fetcher) + self.assertEqual(t.raw_config()["redirects"], {"/2017/old-name/": "/2018/03/02/first-post/"}) + self.assertEqual(len(heads), 2) # probes are cached in _api/probes.json + + def test_paths_cannot_leave_the_source_folder(self): + t = standard_site(self) + with self.assertRaises(wp.WpImportError): + wp.write_file(t.src, "../escape.md", "x", False, False) + self.assertEqual(wp.safe_slug("../../etc/passwd", 7), "etcpasswd") + self.assertEqual(wp.safe_slug("...", 7), "post-7") + + + def test_probe_skips_links_that_cannot_go_into_htaccess(self): + heads = [] + + def head(url): + heads.append(url) + return {WP_URL + "/moved/": (301, WP_URL + "/new%0Aline/"), WP_URL + "/fine-old/": (301, WP_URL + "/fine-new/")}.get(url, (404, None)) + + t = Site(self) + fetcher = wp.Fetcher(WP_URL, t.api, head=head) + redirects, lines = wp.probe_links(fetcher, WP_URL, ["/fine-old/", "/with%20space/", "/line%0Abreak/", "/moved/"], + set(), os.path.join(t.api, "probes.json"), True) + self.assertEqual(redirects, {"/fine-old/": "/fine-new/"}) + self.assertEqual(sorted(heads), [WP_URL + "/fine-old/", WP_URL + "/moved/"]) # unsafe links are not even requested + report = "\n".join(lines) + self.assertIn("with%20space", report) + self.assertIn("line%0Abreak", report) + self.assertIn("skipped", report) + self.assertIn("/moved/ -> '/new%0Aline/': skipped", report) + + def test_comments_file_is_relative_to_the_config_folder_and_the_generator_reads_it(self): + t = standard_site(self, source="src") + t.run() + self.assertTrue(os.path.isfile(os.path.join(t.dir, "comments.json"))) + self.assertFalse(os.path.exists(os.path.join(t.src, "comments.json"))) + if blog.markdown: + out = os.path.join(t.dir, "out") + blog.build(blog.load_config(t.config_path), out) + with open(os.path.join(out, "2018/04/01/second-post/index.html"), encoding="utf-8") as f: + self.assertIn("Nice & good", f.read()) + + def test_percent_encoded_slugs_get_readable_file_names_and_keep_their_url(self): + self.assertEqual(wp.safe_slug("caf%c3%a9-night", 5), "café-night") + self.assertEqual(wp.safe_slug("%2e%2e%2fetc", 5), "etc") + self.assertEqual(wp.safe_slug("%e6%97%a5%e6%9c%ac", 5), "日本") + t = Site(self) + slug = "caf%c3%a9-night" + t.api_data( + posts=[post_json(31, "Café Night", slug, "2018-05-01T09:00:00", f"<p>{words(30, 4)}</p>")], + pages=[{"id": 41, "slug": "d%c3%a9j%c3%a0", "status": "publish", "link": WP_URL + "/d%c3%a9j%c3%a0/", + "title": {"rendered": "Déjà"}, "content": {"rendered": "<p>Page text.</p>", "protected": False}}]) + t.run() + self.assertTrue(os.path.isfile(os.path.join(t.src, "wordpress", "2018-05-01-café-night.md"))) + raw = t.raw_config() + self.assertEqual(raw["posts"]["wordpress/2018-05-01-café-night.md"]["slug"], slug) # the URL is what WordPress has + self.assertEqual(raw["wp_ids"]["31"], f"/2018/05/01/{slug}/") + page = t.read("pages/déjà.md") + self.assertTrue(page.startswith("---\npublished: true\nslug: d%c3%a9j%c3%a0\n---\n")) + blog.load_config(t.config_path) + if blog.markdown: + out = os.path.join(t.dir, "out") + blog.build(blog.load_config(t.config_path), out) + self.assertTrue(wp.output_has(out, f"/2018/05/01/{slug}/")) + self.assertTrue(wp.output_has(out, "/d%c3%a9j%c3%a0/")) + self.assertTrue(os.path.isfile(os.path.join(out, "2018/05/01/café-night/index.html"))) + + +@needs_markdown +class Check(unittest.TestCase): + def build_checked(self, t, **kw): + t.run() + report, problems = wp.run_check(t.config_path, **kw) + return report, problems + + def test_clean_import_passes(self): + t = standard_site(self) + report, problems = self.build_checked(t) + self.assertEqual(problems, 0, report) + self.assertIn("OK", report) + + def test_differences_are_listed(self): + t = standard_site(self) + t.run() + text = t.read("2018-03-02-first-post.md") + t.file("2018-03-02-first-post.md", text.replace(words(60, 1), " ".join(words(60, 1).split()[:50]) + " added words in the middle of it")) + raw = t.raw_config() + raw["redirects"] = {"/old/": "/does/not/exist/"} + t.config = raw + t.save_config() + report, problems = wp.run_check(t.config_path) + self.assertGreaterEqual(problems, 2) + self.assertRegex(report, r"Posts below 0\.98 similarity[^\n]*: 1 ==\n 0\.\d+ /2018/03/02/first-post/") + self.assertIn("/old/ -> /does/not/exist/", report) + + def test_br_counts_and_missing_urls(self): + t = standard_site(self) + t.run() + raw = t.raw_config() + raw["line_breaks"] = "markdown" + t.config = raw + t.save_config() + report, problems = wp.run_check(t.config_path) + self.assertIn("<br> counts that differ: 1", report) + self.assertIn("WordPress 1, built 0", report) + raw["posts"]["wordpress/2018-04-01-second-post.md"]["published"] = False + t.config = raw + t.save_config() + report, _ = wp.run_check(t.config_path) + self.assertIn("post /2018/04/01/second-post/", report) + self.assertIn("Old URLs missing from the output: 1", report) + + def test_check_needs_the_cache(self): + t = Site(self) + with self.assertRaises(wp.WpImportError): + wp.run_check(t.config_path) + + +if __name__ == "__main__": + unittest.main()