|
#!/usr/bin/env bash |
|
set -euo pipefail |
|
|
|
NB_API="https://api.nb.no" |
|
# Default: 1908 "Eventyrbok for børn" by Asbjørnsen & Moe — public domain, |
|
# accessible worldwide, good OCR quality. |
|
NB_ITEM_ID="1747e740ed9158e27927b48abf9c8e43" |
|
OUTPUT_DIR="./folkeeventyr-json" |
|
WP_PATH="" |
|
WP_URL="" |
|
POST_STATUS="draft" |
|
APPLY=0 |
|
LIMIT=0 |
|
DELAY="0.25" |
|
|
|
usage() { |
|
cat <<'USAGE' |
|
Usage: |
|
import-folkeeventyr.sh [options] |
|
|
|
Fetch every folk tale from a digitised book in the National Library of Norway |
|
(api.nb.no), save one JSON file per story, and optionally create or update one |
|
WordPress page per story. |
|
|
|
Text is extracted from IIIF manifests and ALTO OCR XML. The default source is |
|
the 1908 "Eventyrbok for børn" by Asbjørnsen & Moe (public domain, accessible |
|
worldwide). Use --nb-item-id to target a different item. |
|
|
|
Options: |
|
--nb-item-id ID nb.no item ID (default: 1908 Asbjørnsen & Moe collection) |
|
--output-dir DIR JSON destination (default: ./folkeeventyr-json) |
|
--wp-path DIR WordPress root; required with --apply |
|
--wp-url URL Target site URL (recommended; required for multisite) |
|
--status STATUS draft, pending, private, or publish (default: draft) |
|
--delay SECONDS Delay between API requests (default: 0.25) |
|
--limit NUMBER Process only NUMBER stories; useful for testing |
|
--apply Write pages to WordPress (otherwise fetch only) |
|
-h, --help Show help |
|
|
|
Examples: |
|
./import-folkeeventyr.sh --limit 2 |
|
./import-folkeeventyr.sh --output-dir ./stories \ |
|
--wp-path /var/www/html --wp-url https://example.test --apply |
|
|
|
Requirements: bash, curl, python3; additionally wp when using --apply. |
|
USAGE |
|
} |
|
|
|
die() { |
|
printf 'Error: %s\n' "$*" >&2 |
|
exit 1 |
|
} |
|
|
|
log() { |
|
printf '[%s] %s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$*" |
|
} |
|
|
|
while (($#)); do |
|
case "$1" in |
|
--nb-item-id) |
|
(($# >= 2)) || die "--nb-item-id needs a value" |
|
NB_ITEM_ID=$2 |
|
shift 2 |
|
;; |
|
--output-dir) |
|
(($# >= 2)) || die "--output-dir needs a value" |
|
OUTPUT_DIR=$2 |
|
shift 2 |
|
;; |
|
--wp-path) |
|
(($# >= 2)) || die "--wp-path needs a value" |
|
WP_PATH=$2 |
|
shift 2 |
|
;; |
|
--wp-url) |
|
(($# >= 2)) || die "--wp-url needs a value" |
|
WP_URL=$2 |
|
shift 2 |
|
;; |
|
--status) |
|
(($# >= 2)) || die "--status needs a value" |
|
POST_STATUS=$2 |
|
shift 2 |
|
;; |
|
--delay) |
|
(($# >= 2)) || die "--delay needs a value" |
|
DELAY=$2 |
|
shift 2 |
|
;; |
|
--limit) |
|
(($# >= 2)) || die "--limit needs a value" |
|
LIMIT=$2 |
|
shift 2 |
|
;; |
|
--apply) |
|
APPLY=1 |
|
shift |
|
;; |
|
-h|--help) |
|
usage |
|
exit 0 |
|
;; |
|
*) |
|
die "Unknown option: $1" |
|
;; |
|
esac |
|
done |
|
|
|
[[ $POST_STATUS =~ ^(draft|pending|private|publish)$ ]] || |
|
die "Invalid --status: $POST_STATUS" |
|
[[ $LIMIT =~ ^[0-9]+$ ]] || die "--limit must be a non-negative integer" |
|
[[ $DELAY =~ ^[0-9]+([.][0-9]+)?$ ]] || |
|
die "--delay must be a non-negative number" |
|
|
|
for command_name in curl python3; do |
|
command -v "$command_name" >/dev/null || |
|
die "Required command not found: $command_name" |
|
done |
|
|
|
if ((APPLY)); then |
|
command -v wp >/dev/null || die "Required command not found: wp" |
|
[[ -n $WP_PATH ]] || die "--wp-path is required with --apply" |
|
[[ -f "$WP_PATH/wp-config.php" ]] || |
|
die "No wp-config.php found under --wp-path: $WP_PATH" |
|
fi |
|
|
|
mkdir -p "$OUTPUT_DIR" |
|
OUTPUT_DIR=$(cd "$OUTPUT_DIR" && pwd) |
|
WORK_DIR=$(mktemp -d "${TMPDIR:-/tmp}/folkeeventyr.XXXXXX") |
|
trap 'rm -rf "$WORK_DIR"' EXIT |
|
|
|
SLUG_LIST="$WORK_DIR/slugs.txt" |
|
|
|
log "Fetching folk tales from nb.no (item: $NB_ITEM_ID)" |
|
|
|
python3 - "$NB_ITEM_ID" "$NB_API" "$OUTPUT_DIR" "$WORK_DIR" "$LIMIT" "$DELAY" "$SLUG_LIST" <<'PY' |
|
import json, os, re, sys, time, xml.etree.ElementTree as ET |
|
from datetime import datetime, timezone |
|
from html import escape |
|
from urllib.request import urlopen, Request |
|
from urllib.error import URLError, HTTPError |
|
|
|
item_id, nb_api, output_dir, work_dir, limit_str, delay_str, slug_list_file = sys.argv[1:8] |
|
limit = int(limit_str) |
|
delay = float(delay_str) |
|
|
|
|
|
def fetch(url, as_text=False): |
|
req = Request(url, headers={"User-Agent": "folkeeventyr-nb-importer/1.0"}) |
|
with urlopen(req) as r: |
|
raw = r.read() |
|
return raw.decode("utf-8", errors="replace") if as_text else json.loads(raw) |
|
|
|
|
|
def slugify(title): |
|
s = title.lower() |
|
for src, dst in [("æ", "ae"), ("ø", "o"), ("å", "aa"), ("é", "e"), |
|
("è", "e"), ("ü", "y"), ("ö", "o"), ("ä", "a")]: |
|
s = s.replace(src, dst) |
|
s = re.sub(r"[^\w]", "_", s) |
|
s = re.sub(r"_+", "_", s) |
|
return s.strip("_") |
|
|
|
|
|
def strip_xmlns(xml_str): |
|
"""Remove XML namespaces so ElementTree tags and attributes are unqualified.""" |
|
# Remove namespace declarations: xmlns="..." or xmlns:foo="..." |
|
xml_str = re.sub(r'\s+xmlns(?::\w+)?="[^"]*"', "", xml_str) |
|
# Remove prefixed attributes that would become unbound: xsi:foo="..." xlink:href="..." |
|
xml_str = re.sub(r'\s+\w+:\w+="[^"]*"', "", xml_str) |
|
# Remove namespace prefixes from element names: <foo:bar> → <bar> |
|
xml_str = re.sub(r"<(/?)(\w+):", r"<\1", xml_str) |
|
return xml_str |
|
|
|
|
|
def parse_alto(xml_str): |
|
""" |
|
Parse ALTO XML into a list of text blocks. |
|
Each block: {text, fontsize} |
|
Handles both ALTO 1.2 (no namespace) and ALTO v3 (with namespace). |
|
""" |
|
try: |
|
root = ET.fromstring(strip_xmlns(xml_str)) |
|
except ET.ParseError: |
|
return [] |
|
|
|
styles = {} |
|
for style in root.findall(".//TextStyle"): |
|
sid = style.get("ID", "") |
|
try: |
|
styles[sid] = float(style.get("FONTSIZE", "0")) |
|
except ValueError: |
|
styles[sid] = 0.0 |
|
|
|
blocks = [] |
|
for block in root.findall(".//TextBlock"): |
|
srefs = block.get("STYLEREFS", "") |
|
lines = [] |
|
for line in block.findall(".//TextLine"): |
|
words = [] |
|
prev_was_hyp = False |
|
for elem in line: |
|
tag = elem.tag |
|
if tag == "String": |
|
content = elem.get("CONTENT", "") |
|
if content: |
|
if prev_was_hyp and words: |
|
words[-1] = words[-1] + content |
|
else: |
|
words.append(content) |
|
prev_was_hyp = False |
|
elif tag == "HYP": |
|
prev_was_hyp = True |
|
if words: |
|
lines.append(" ".join(words)) |
|
|
|
text = " ".join(lines).strip() |
|
if not text: |
|
continue |
|
|
|
max_fs = max( |
|
(styles[sid] for sid in srefs.split() if sid in styles), |
|
default=0.0, |
|
) |
|
blocks.append({"text": text, "fontsize": max_fs}) |
|
|
|
return blocks |
|
|
|
|
|
def is_junk(text): |
|
"""True for page numbers, Roman numerals, ornaments, and very short noise.""" |
|
t = text.strip() |
|
return ( |
|
not t |
|
or re.match(r"^[\d\s\-–\.]+$", t) |
|
or re.match(r"^–?\s*[IVXLCDM]+\s*–?$", t) |
|
or re.match(r"^\*+$", t) |
|
or len(t) <= 2 |
|
) |
|
|
|
|
|
def is_story_title(text): |
|
""" |
|
Heuristic: does a large-font block look like a genuine folk tale title? |
|
Rejects: OCR noise, all-uppercase running headers, lowercase fragments. |
|
""" |
|
t = text.strip() |
|
if len(t) <= 2: |
|
return False |
|
# Must be mostly letter characters (OCR noise has punctuation/garbage) |
|
letters = sum(1 for c in t if c.isalpha()) |
|
if not letters or letters / len(t) < 0.65: |
|
return False |
|
# All-uppercase = running header or book/section title, not a story title |
|
if t == t.upper() and not t.islower(): |
|
return False |
|
# Story titles start with an uppercase letter |
|
if not t[0].isupper(): |
|
return False |
|
return True |
|
|
|
|
|
def classify_blocks(blocks): |
|
"""Return (headings, paragraphs) boolean lists.""" |
|
if not blocks: |
|
return [], [] |
|
|
|
font_sizes = [b["fontsize"] for b in blocks if b["fontsize"] > 0] |
|
all_same = len(set(font_sizes)) <= 1 |
|
max_fs = max(font_sizes, default=0) |
|
|
|
headings = [] |
|
paragraphs = [] |
|
for b in blocks: |
|
text = b["text"] |
|
if is_junk(text): |
|
headings.append(False) |
|
paragraphs.append(False) |
|
continue |
|
if not all_same and b["fontsize"] == max_fs and len(text) < 80 and is_story_title(text): |
|
headings.append(True) |
|
paragraphs.append(False) |
|
else: |
|
headings.append(False) |
|
paragraphs.append(True) |
|
|
|
return headings, paragraphs |
|
|
|
|
|
# ── Fetch item metadata ────────────────────────────────────────────────────── |
|
print(f"Fetching metadata for item {item_id} …", file=sys.stderr) |
|
try: |
|
item_data = fetch(f"{nb_api}/catalog/v1/items/{item_id}") |
|
except (URLError, HTTPError) as exc: |
|
raise SystemExit(f"Could not fetch item metadata from nb.no: {exc}") |
|
|
|
meta = item_data.get("metadata", {}) |
|
item_title = meta.get("title") or "Norske Folkeeventyr" |
|
item_url = f"https://www.nb.no/items/{item_id}" |
|
print(f"Source: {item_title} ({item_url})", file=sys.stderr) |
|
|
|
# ── Fetch IIIF manifest ────────────────────────────────────────────────────── |
|
print("Fetching IIIF manifest …", file=sys.stderr) |
|
try: |
|
manifest = fetch(f"{nb_api}/catalog/v1/iiif/{item_id}/manifest") |
|
except (URLError, HTTPError) as exc: |
|
raise SystemExit(f"Could not fetch IIIF manifest from nb.no: {exc}") |
|
|
|
canvases = manifest["sequences"][0]["canvases"] |
|
print(f"Manifest contains {len(canvases)} pages", file=sys.stderr) |
|
|
|
# ── Process pages ──────────────────────────────────────────────────────────── |
|
current_title = None |
|
current_paragraphs = [] |
|
stories_saved = [] |
|
used_slugs = set() |
|
|
|
|
|
def flush_story(): |
|
if not current_title or not current_paragraphs: |
|
return None |
|
|
|
slug = slugify(current_title) |
|
if not slug: |
|
return None |
|
|
|
base = slug |
|
n = 2 |
|
while slug in used_slugs: |
|
slug = f"{base}_{n}" |
|
n += 1 |
|
used_slugs.add(slug) |
|
|
|
content = "\n\n".join( |
|
f"<p>{escape(p, quote=False)}</p>" for p in current_paragraphs |
|
) |
|
record = { |
|
"title": current_title, |
|
"slug": slug, |
|
"content": content, |
|
"source_url": item_url, |
|
"fetched_at": datetime.now(timezone.utc).isoformat(), |
|
} |
|
|
|
json_path = os.path.join(output_dir, f"{slug}.json") |
|
content_path = os.path.join(work_dir, f"{slug}-content.html") |
|
|
|
with open(json_path, "w", encoding="utf-8") as fh: |
|
json.dump(record, fh, ensure_ascii=False, indent=2) |
|
fh.write("\n") |
|
with open(content_path, "w", encoding="utf-8") as fh: |
|
fh.write(content + "\n") |
|
|
|
stories_saved.append(slug) |
|
return slug |
|
|
|
|
|
for canvas in canvases: |
|
if limit > 0 and len(stories_saved) >= limit: |
|
break |
|
|
|
# Extract page URN from the @seeAlso Alto link |
|
see_also = canvas.get("@seeAlso", {}) |
|
alto_href = (see_also.get("@id", "") if isinstance(see_also, dict) |
|
else next((x.get("@id", "") for x in see_also |
|
if x.get("format", "").endswith("alto+xml")), "") |
|
if isinstance(see_also, list) else "") |
|
if not alto_href: |
|
continue |
|
|
|
page_urn = alto_href.rsplit("/", 1)[-1] |
|
label = canvas.get("label", "") |
|
|
|
# Skip front-matter: covers (-), blank pages, and Roman-numeral preface pages |
|
if label == "-" or re.match(r"^[IVXLCDM]+$", label or ""): |
|
continue |
|
|
|
alto_url = f"{nb_api}/catalog/v1/metadata/{item_id}/altos/{page_urn}" |
|
try: |
|
xml_str = fetch(alto_url, as_text=True) |
|
except (URLError, HTTPError) as exc: |
|
print(f" Warning: could not fetch ALTO for {page_urn}: {exc}", |
|
file=sys.stderr) |
|
continue |
|
|
|
blocks = parse_alto(xml_str) |
|
headings, paragraphs = classify_blocks(blocks) |
|
|
|
for block, is_heading, is_para in zip(blocks, headings, paragraphs): |
|
if is_heading: |
|
slug = flush_story() |
|
if slug: |
|
print(f" Saved: {current_title}", file=sys.stderr) |
|
if limit > 0 and len(stories_saved) >= limit: |
|
break |
|
current_title = block["text"] |
|
current_paragraphs = [] |
|
elif is_para and current_title: |
|
current_paragraphs.append(block["text"]) |
|
else: |
|
if delay > 0: |
|
time.sleep(delay) |
|
continue |
|
break # inner break propagates here when limit reached |
|
|
|
# Flush final story |
|
if not (limit > 0 and len(stories_saved) >= limit): |
|
slug = flush_story() |
|
if slug: |
|
print(f" Saved: {current_title}", file=sys.stderr) |
|
|
|
print(f"Extracted {len(stories_saved)} stories", file=sys.stderr) |
|
|
|
with open(slug_list_file, "w", encoding="utf-8") as fh: |
|
fh.write("\n".join(stories_saved) + ("\n" if stories_saved else "")) |
|
PY |
|
|
|
mapfile -t STORY_SLUGS < "$SLUG_LIST" |
|
log "Found ${#STORY_SLUGS[@]} stories" |
|
|
|
WP_ARGS=(--path="$WP_PATH") |
|
if [[ -n $WP_URL ]]; then |
|
WP_ARGS+=(--url="$WP_URL") |
|
fi |
|
|
|
created=0 |
|
updated=0 |
|
scraped=0 |
|
|
|
for slug in "${STORY_SLUGS[@]}"; do |
|
[[ $slug =~ ^[a-zA-Z0-9_-]+$ ]] || |
|
die "Unsafe slug from nb.no extraction: $slug" |
|
|
|
json_file="$OUTPUT_DIR/$slug.json" |
|
content_file="$WORK_DIR/$slug-content.html" |
|
|
|
((scraped += 1)) |
|
|
|
if ((APPLY)); then |
|
title=$(python3 -c \ |
|
'import json,sys; print(json.load(open(sys.argv[1], encoding="utf-8"))["title"])' \ |
|
"$json_file") |
|
|
|
existing_id=$(wp "${WP_ARGS[@]}" post list \ |
|
--post_type=page \ |
|
--post_status=any \ |
|
--meta_key=_folkeeventyr_slug \ |
|
--meta_value="$slug" \ |
|
--field=ID \ |
|
--format=ids) |
|
|
|
if [[ -n $existing_id ]]; then |
|
read -r existing_id _ <<< "$existing_id" |
|
log "Updating page $existing_id: $title" |
|
wp "${WP_ARGS[@]}" post update "$existing_id" "$content_file" \ |
|
--post_title="$title" \ |
|
--post_name="$slug" \ |
|
--post_status="$POST_STATUS" >/dev/null |
|
((updated += 1)) |
|
else |
|
log "Creating page: $title" |
|
post_id=$(wp "${WP_ARGS[@]}" post create "$content_file" \ |
|
--post_type=page \ |
|
--post_title="$title" \ |
|
--post_name="$slug" \ |
|
--post_status="$POST_STATUS" \ |
|
--porcelain) |
|
wp "${WP_ARGS[@]}" post meta update \ |
|
"$post_id" _folkeeventyr_slug "$slug" >/dev/null |
|
((created += 1)) |
|
fi |
|
fi |
|
done |
|
|
|
log "Done: $scraped JSON files written to $OUTPUT_DIR" |
|
if ((APPLY)); then |
|
log "WordPress: $created pages created, $updated pages updated ($POST_STATUS)" |
|
else |
|
log "WordPress unchanged. Re-run with --wp-path and --apply to import." |
|
fi |
