#!/usr/bin/env python3
"""
Full-scale vocabulary database builder for Rijksmuseum MCP+.

Builds a SQLite database mapping vocabulary terms (Iconclass subjects, depicted
persons/places, materials, techniques, types, creators) to artworks, enabling
vocabulary-based search in the MCP server.

Phases:
  0.  Parse ALL data dumps (classification, person, place, concept, event, topical_term, organisation)
  0.5 Seed curated set names from OAI-PMH ListSets
  1.  Harvest ALL OAI-PMH records (836K+), extract 9 vocabulary fields + Linked Art URIs
  2.  Resolve unmatched vocabulary URIs via Linked Art API (multi-threaded)
  4.  Resolve all artwork Linked Art URIs for Tier 2 fields: inscriptions, provenance,
      credit lines, dimensions, production roles, attribution qualifiers (multi-threaded)
  2b. Re-resolve vocabulary URIs introduced by Phase 4 (production roles, attribution qualifiers)
  3.  Post-processing: geocoding import, normalized labels, FTS5 indexes, validation stats (runs last)

Usage:
    python3 scripts/harvest-vocabulary-db.py                # Full run (all phases)
    python3 scripts/harvest-vocabulary-db.py --resume       # Resume from checkpoint
    python3 scripts/harvest-vocabulary-db.py --skip-dump    # Skip Phase 0 (no local dump)
    python3 scripts/harvest-vocabulary-db.py --phase 3      # Run only from phase N onward
    python3 scripts/harvest-vocabulary-db.py --phase 4      # Run Phase 4 + 3 only
    python3 scripts/harvest-vocabulary-db.py --threads 16   # Phase 4 thread count
    python3 scripts/harvest-vocabulary-db.py --phase 3 --geo-csv data/backfills/geocoded-places.csv
    python3 scripts/harvest-vocabulary-db.py --limit 1   # Test harvest: 1 page (~200 artworks), all phases

Output: data/vocabulary.db (~1 GB)
"""

import argparse
import csv
import hashlib
import html
import json
import math
import os
import re
import socket
import sqlite3
import subprocess
import zipfile
from collections import Counter, defaultdict
import sys
import tarfile
from itertools import chain
import time
import urllib.error
import urllib.parse
import urllib.request
import requests
import requests.adapters
import xml.etree.ElementTree as ET
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from pathlib import Path

# Force line-buffered stdout/stderr so progress prints appear in `tee` output
# immediately, regardless of how the interpreter was invoked. See CLAUDE.md
# "Python stdout buffering when piped to `tee`" — caused an hour of "is it
# stuck?" confusion on the 2026-04-17 recovery run.
sys.stdout.reconfigure(line_buffering=True)
sys.stderr.reconfigure(line_buffering=True)

# ─── Configuration ───────────────────────────────────────────────────

SCRIPT_DIR = Path(__file__).parent
PROJECT_DIR = SCRIPT_DIR.parent

# Make scripts/lib importable as a namespace package (#222 audit module).
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))
from lib.harvest_audit import (  # noqa: E402
    AuditResult,
    final_summary,
    format_stdout_table,
    run_phase_audit,
)
from lib import enrichment_methods as em  # noqa: E402

HARVEST_VERSION = "v0.24"
DB_PATH = PROJECT_DIR / "data" / "vocabulary.db"
CHECKPOINT_PATH = SCRIPT_DIR / ".harvest-checkpoint"
DUMPS_DIR = Path.home() / "Downloads" / "rijksmuseum-data-dumps"


# ─── Run state (#230) ────────────────────────────────────────────────
# Per-phase wall durations (seconds) keyed `phase0`, `phase0_5`, `phase1`,
# `phase2`, `phase4`, `phase4_5`, `phase2b`, `phase3`. Populated by main();
# persisted to version_info as `<key>_duration_sec`. `--start-phase N` resumes
# only stamp the phases that actually ran — keys for skipped phases are absent.
PHASE_DURATIONS: dict[str, float] = {}
HARVEST_STARTED_AT: str | None = None  # ISO-8601 UTC


def repo_path(p) -> str:
    """#230: Render a path as repo-relative when it lives under PROJECT_DIR,
    falling back to the absolute string otherwise. Use for log lines so logs
    captured on the build machine remain interpretable when read elsewhere."""
    try:
        return str(Path(p).resolve().relative_to(PROJECT_DIR.resolve()))
    except (ValueError, OSError):
        return str(p)


def detect_git_commit() -> tuple[str, bool]:
    """#230: Return (short_commit, dirty) for the harvest script's checkout.
    Returns ("unknown", False) outside a git checkout (e.g. Railway runtime)."""
    try:
        commit = subprocess.check_output(
            ["git", "rev-parse", "HEAD"],
            cwd=PROJECT_DIR,
            stderr=subprocess.DEVNULL,
            timeout=5,
        ).decode().strip()
        dirty = subprocess.call(
            ["git", "diff", "--quiet"],
            cwd=PROJECT_DIR,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=5,
        ) != 0
        return commit[:8], dirty
    except (subprocess.SubprocessError, FileNotFoundError, OSError):
        return "unknown", False

# ── Enrichment dump artefacts (#242 part 3) ──
# Phase 3 enrichment (folded in from scripts/enrich-vocab-from-dumps.py) reads
# these four artefacts from DUMPS_DIR to populate person life data, Wikidata
# links, place hierarchy, and concept hierarchy. When any artefact is missing,
# the corresponding enrichment phase is skipped with a warning — the DB still
# opens because the column stubs are added unconditionally in run_phase3().
ENRICH_EDM_ACTORS_ZIP = "201911-rma-edm-actors.zip"
ENRICH_SKOS_THESAURUS_ZIP = "201911-rma-skos-thesaurus.zip"
ENRICH_PLACE_DIR = "place_extracted"
ENRICH_PERSON_DIR = "person_extracted"

# Dump files to parse in Phase 0, with their vocabulary type mappings.
# NOTE on ID namespaces:
#   - classification (22xxx) → matches dc:subject Iconclass refs directly
#   - concept (12xxx) → matches dcterms:medium, dc:type, dcterms:technique refs
#   - person (31xxx) → matches dc:creator refs (bibliographic/artist persons)
#     BUT NOT depicted persons (21xxx) — those need Phase 2 HTTP resolution
#   - place (13xxx) → matches dcterms:spatial refs (production places)
#     BUT NOT depicted places (23xxx) — those need Phase 2 HTTP resolution
#   - event (24xxx) → partially matches dc:subject event refs
DUMP_CONFIGS = [
    # (tar.gz name, default vocab type)
    ("classification", "classification"),
    ("concept", "classification"),
    ("topical_term", "classification"),
    ("person", "person"),
    ("organisation", "organisation"),  # #238: was "person", incorrect for Schema.org org dumps
    ("place", "place"),
    ("event", "event"),
    ("exhibition", "event"),
]

OAI_BASE = "https://data.rijksmuseum.nl/oai"
LINKED_ART_BASE = "https://data.rijksmuseum.nl"
USER_AGENT = "rijksmuseum-mcp-harvest/1.0"
DEFAULT_THREADS = 12
BATCH_SIZE = 500  # Commit every N pages

# Module-level HTTP session with a connection pool shared across threads.
# Amortises TCP+TLS handshake cost (~90 ms/request) across the Phase 4 worker
# pool — the previous urllib.request-based path opened a fresh connection for
# every artwork, pinning ~9 cores on TLS crypto and capping throughput at ~5/s.
# Pool sized for 32 concurrent workers (generous headroom over DEFAULT_THREADS).
# Callers handle their own retries (_resolve_one, resolve_artwork have distinct
# error-categorisation semantics), so max_retries=0 on the adapter.
_HTTP_SESSION: requests.Session | None = None


def get_http_session() -> requests.Session:
    global _HTTP_SESSION
    if _HTTP_SESSION is None:
        s = requests.Session()
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=32,
            pool_maxsize=32,
            max_retries=0,
        )
        s.mount("https://", adapter)
        s.mount("http://", adapter)
        _HTTP_SESSION = s
    return _HTTP_SESSION

# ─── AAT URIs for Tier 2 Linked Art parsing ─────────────────────────

AAT_INSCRIPTIONS = "http://vocab.getty.edu/aat/300435414"
AAT_PROVENANCE = "http://vocab.getty.edu/aat/300444174"
AAT_CREDIT_LINE = "http://vocab.getty.edu/aat/300026687"
AAT_UNIT_CM = "http://vocab.getty.edu/aat/300379098"
AAT_UNIT_MM = "http://vocab.getty.edu/aat/300379097"
AAT_UNIT_M = "http://vocab.getty.edu/aat/300379100"
AAT_HEIGHT = "http://vocab.getty.edu/aat/300055644"
AAT_WIDTH = "http://vocab.getty.edu/aat/300055647"
RM_HEIGHT = "https://id.rijksmuseum.nl/22011"
RM_WIDTH = "https://id.rijksmuseum.nl/22012"
HEIGHT_URIS = {AAT_HEIGHT, RM_HEIGHT}
WIDTH_URIS = {AAT_WIDTH, RM_WIDTH}
AAT_NARRATIVE = "http://vocab.getty.edu/aat/300048722"
AAT_DESCRIPTION = "http://vocab.getty.edu/aat/300435452"
AAT_PRODUCTION_STATEMENT = "http://vocab.getty.edu/aat/300435416"
AAT_DEPTH = "http://vocab.getty.edu/aat/300072633"
AAT_WEIGHT = "http://vocab.getty.edu/aat/300056240"
AAT_DIAMETER = "http://vocab.getty.edu/aat/300055624"
RM_DEPTH = "https://id.rijksmuseum.nl/2203"
RM_WEIGHT = "https://id.rijksmuseum.nl/220217"
RM_DIAMETER = "https://id.rijksmuseum.nl/2205120"
DEPTH_URIS = {AAT_DEPTH, RM_DEPTH}
WEIGHT_URIS = {AAT_WEIGHT, RM_WEIGHT}
DIAMETER_URIS = {AAT_DIAMETER, RM_DIAMETER}
AAT_UNIT_G = "http://vocab.getty.edu/aat/300379225"     # grams
AAT_UNIT_KG = "http://vocab.getty.edu/aat/300379226"    # kilograms

# Source object types (used_object_of_type) — AAT concepts for the type of object
# used as a source in production (e.g., "this print was made after a painting").
# Only 6 distinct values across the collection; all are external AAT URIs not in the
# Rijksmuseum vocabulary dump, so they must be seeded into the vocabulary table.
AAT_SOURCE_TYPES = {
    "300102051": "designs",
    "300033973": "drawings",
    "300033618": "paintings",
    "300041273": "prints (visual works)",
    "300047090": "sculpture (visual works)",
    "300046300": "photographs",
}

# Exhibition N-Triples predicates
P_BEGIN = "http://www.cidoc-crm.org/cidoc-crm/P82a_begin_of_the_begin"
P_END = "http://www.cidoc-crm.org/cidoc-crm/P82b_end_of_the_end"
P_HAS_MEMBER = "http://www.cidoc-crm.org/cidoc-crm/P46i_forms_part_of"  # legacy fallback: artwork → exhibition
P_COMPOSED_OF = "http://www.cidoc-crm.org/cidoc-crm/P46_is_composed_of"  # legacy fallback: exhibition → artwork (direct)
P_HAS_MEMBER_LA = "https://linked.art/ns/terms/has_member"              # actual predicate in exhibition dump
P_USED_SPECIFIC_OBJECT = "http://www.cidoc-crm.org/cidoc-crm/P16_used_specific_object"  # exhibition → Set bnode
P_IS_IDENTIFIED_BY = "http://www.cidoc-crm.org/cidoc-crm/P1_is_identified_by"  # exhibition → title bnode
P_HAS_TIME_SPAN = "http://www.cidoc-crm.org/cidoc-crm/P4_has_time-span"        # exhibition → timespan bnode

# Title classification AAT URIs
AAT_TITLE_FULL = "http://vocab.getty.edu/aat/300417200"
AAT_TITLE_BRIEF = "http://vocab.getty.edu/aat/300417207"
AAT_TITLE_DISPLAY = "http://vocab.getty.edu/aat/300404670"
RM_TITLE_FORMER = "https://id.rijksmuseum.nl/22015528"

# ─── N-Triples parsing (same as pilot) ──────────────────────────────

_LITERAL_BODY = r'(?:[^"\\]|\\.)*'  # any char except `"`/`\`, OR a `\`-escape pair

NT_PATTERN = re.compile(
    # Object can be <uri> or "literal" optionally followed by a datatype
    # suffix ^^<datatype> (e.g. xsd:dateTime on P82a/P82b literals).
    # The datatype group is non-capturing — we don't use it.
    # The literal body uses `_LITERAL_BODY` so escaped quotes (`\"`) and
    # backslashes (`\\`) inside literals don't terminate the match early
    # (#245: ~143 organisation files had escaped-quote names like
    # `Verein \"Arbeitsgruppe ...\"` and were being silently dropped).
    r'^<([^>]+)>\s+<([^>]+)>\s+(?:<([^>]+)>|"(' + _LITERAL_BODY + r')"(?:\^\^<[^>]+>)?)\s*\.\s*$'
)
BNODE_PATTERN = re.compile(
    r'^_:(\S+)\s+<([^>]+)>\s+(?:<([^>]+)>|"(' + _LITERAL_BODY + r')"(?:\^\^<[^>]+>)?)\s*\.\s*$'
)
# Matches triples with URI subject and blank-node object: <uri> <uri> _:bnode .
# Needed for exhibition dump parsing where <Exhibition> P16_used_specific_object _:Set .
# NT_PATTERN's object alternation only accepts URIs/literals, so these triples
# are invisible to it — without this pattern, the exhibition → Set link is lost
# and has_member edges cannot be attributed to the right exhibition.
NT_TO_BNODE_PATTERN = re.compile(
    r'^<([^>]+)>\s+<([^>]+)>\s+_:(\S+)\s*\.\s*$'
)
NT_LANG_PATTERN = re.compile(
    # Language tag follows BCP 47: primary subtag plus optional region/script
    # subtags separated by hyphens (e.g. `@nl`, `@nl-NL`, `@zh-Hans-CN`).
    # The earlier `\w+` form silently dropped topical_term `@nl-NL` labels,
    # which contributed to the Schema.org dump parsing gap (#238).
    r'^<([^>]+)>\s+<([^>]+)>\s+"(' + _LITERAL_BODY + r')"@([A-Za-z][A-Za-z0-9-]*)\s*\.\s*$'
)


# N-Triples literal escape sequences: \" → ", \\ → \, \n → newline, etc.
# https://www.w3.org/TR/n-triples/#grammar-production-ECHAR
_ECHAR_MAP = {
    '"': '"', "\\": "\\", "n": "\n", "r": "\r", "t": "\t",
    "b": "\b", "f": "\f", "/": "/",
}


def _unescape_nt_literal(s: str) -> str:
    """Decode N-Triples escape sequences in a literal body. The regex captures
    the raw bytes between quotes; this turns `\\"` back into `"`, `\\\\` into
    `\\`, `\\n` into a newline, etc. Unknown escapes pass through as-is so we
    don't lose data on novel sequences."""
    if "\\" not in s:
        return s
    out: list[str] = []
    i = 0
    n = len(s)
    while i < n:
        c = s[i]
        if c == "\\" and i + 1 < n:
            nxt = s[i + 1]
            out.append(_ECHAR_MAP.get(nxt, nxt))
            i += 2
        else:
            out.append(c)
            i += 1
    return "".join(out)

P_LABEL = "http://www.cidoc-crm.org/cidoc-crm/P190_has_symbolic_content"
P_LANGUAGE = "http://www.cidoc-crm.org/cidoc-crm/P72_has_language"
P_EQUIVALENT = "https://linked.art/ns/terms/equivalent"
SKOS_PREFLABEL = "http://www.w3.org/2004/02/skos/core#prefLabel"
RDFS_LABEL = "http://www.w3.org/2000/01/rdf-schema#label"
P_BROADER = "http://www.w3.org/2004/02/skos/core#broader"
P_HAS_TYPE = "http://www.cidoc-crm.org/cidoc-crm/P2_has_type"
LANG_EN = "http://vocab.getty.edu/aat/300388277"
LANG_NL = "http://vocab.getty.edu/aat/300388256"
AAT_DISPLAY_NAME = AAT_TITLE_DISPLAY  # same AAT concept (300404670)
AAT_PREFERRED_NAME = "http://vocab.getty.edu/aat/300404671"
AAT_INVERTED_NAME = "http://vocab.getty.edu/aat/300404672"

# AAT name classification URI suffix → label (for person_names table)
AAT_NAME_CLASSIFICATION = {
    "300404670": "display",
    "300404671": "preferred",
    "300404672": "inverted",
}

# Linked Art type → vocabulary type
LA_TYPE_MAP = {
    "Person": "person",
    "Group": "group",
    "Actor": "person",
    "Place": "place",
    "Activity": "event",
    "Set": "classification",
    "Type": "classification",
    "Material": "classification",
    "MeasurementUnit": "classification",
    "Language": "classification",
    "Currency": "classification",
}

# ─── Schema.org dump constants (#238) ───────────────────────────────
#
# The person / organisation / topical_term dumps use Schema.org shape
# (`schema:name` directly on the entity, `schema:sameAs` for external
# authority links, `rdf:type schema:Person|Organization|DefinedTerm`).
# The place dump is a hybrid: ~12% Schema.org shape, ~88% Linked Art.
# Before #238, these were silently dropped by parse_nt_file because the
# function only handled Linked Art / CIDOC-CRM with blank-node appellations.

P_SCHEMA_NAME = "http://schema.org/name"
P_SCHEMA_ALTERNATE_NAME = "http://schema.org/alternateName"
P_SCHEMA_SAME_AS = "http://schema.org/sameAs"
RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"

# Schema.org rdf:type values that mark an entity as "Schema.org shape".
# Place is included because the place dump is a hybrid — see #238.
SCHEMA_TYPE_MAP = {
    "http://schema.org/Person":       "person",
    "http://schema.org/Organization": "organisation",
    "http://schema.org/DefinedTerm":  "classification",
    "http://schema.org/Place":        "place",
}


# (substring, authority) pairs — order matters only for efficiency (more
# common first). Both sws.geonames.org and www.geonames.org fold into
# "geonames" via the shared `geonames.org/` needle.
_AUTHORITY_NEEDLES: tuple[tuple[str, str], ...] = (
    ("wikidata.org/entity/",   "wikidata"),
    ("viaf.org/viaf/",         "viaf"),
    ("rkd.nl",                 "rkd"),
    ("vocab.getty.edu/ulan/",  "ulan"),
    ("vocab.getty.edu/tgn/",   "tgn"),
    ("vocab.getty.edu/aat/",   "aat"),
    ("iconclass.org/",         "iconclass"),
    ("geonames.org/",          "geonames"),
    ("biografischportaal.nl",  "biografisch_portaal"),
    ("cerl.org",               "cerl"),
    ("pic.nypl.org",           "nypl"),
    # The rijks_internal needle is intentionally last — anything matching a
    # genuinely external authority above takes precedence.
    ("hdl.handle.net/",        "handle"),
    ("id.rijksmuseum.nl/",     "rijks_internal"),
)


# Charset guard against upstream sources wrapping non-conforming IDs
# inside an authority's URI shape. Catches parse-time mismatches only;
# well-formed-but-nonexistent IDs are caught at lookup time.
_NUMERIC_ID = re.compile(r"^\d+$")
_AUTHORITY_ID_PATTERNS: dict[str, re.Pattern[str]] = {
    "wikidata":  re.compile(r"^Q\d+$"),
    "iconclass": re.compile(r"^\d+[A-Z]"),
    "viaf":      _NUMERIC_ID,
    "ulan":      _NUMERIC_ID,
    "tgn":       _NUMERIC_ID,
    "aat":       _NUMERIC_ID,
    "geonames":  _NUMERIC_ID,
}


def classify_authority(uri: str) -> tuple[str, str]:
    """Classify an external sameAs URI into (authority, local_id).

    Covers all authorities observed in the Nov 2024 data dumps across
    person / organisation / topical_term / place (see #238 sweep).
    Unknown authorities — and host-matched URIs whose local_id fails the
    per-authority format check — fall through to ('other', uri).
    """
    if not uri:
        return ("other", uri)
    # Normalize trailing-slash and no-slash variants so `viaf/19908139/`
    # and `viaf/19908139` both yield `19908139`.
    u = uri.rstrip("/")
    for needle, authority in _AUTHORITY_NEEDLES:
        if needle in u:
            local_id = u.rsplit("/", 1)[-1]
            pat = _AUTHORITY_ID_PATTERNS.get(authority)
            if pat is not None and not pat.match(local_id):
                return ("other", uri)
            return (authority, local_id)
    return ("other", uri)


# External vocabulary IDs (e.g. Getty AAT) used directly in OAI-PMH dc:type.
# The Rijksmuseum API returns 404 for these since they aren't Rijksmuseum entities.
EXTERNAL_VOCAB = {
    "300078817": {"type": "classification", "label_en": "rectos", "label_nl": "rectozijden", "external_id": "http://vocab.getty.edu/aat/300078817"},
    "300010292": {"type": "classification", "label_en": "versos", "label_nl": "versozijden", "external_id": "http://vocab.getty.edu/aat/300010292"},
    "300404450": {"type": "classification", "label_en": "primary", "label_nl": "primair", "external_id": "http://vocab.getty.edu/aat/300404450"},
    "300404451": {"type": "classification", "label_en": "secondary", "label_nl": "secundair", "external_id": "http://vocab.getty.edu/aat/300404451"},
    "300379012": {"type": "classification", "label_en": "undetermined", "label_nl": "onbepaald", "external_id": "http://vocab.getty.edu/aat/300379012"},
    # Attribution qualifiers — rich labels for assigned_by[].classified_as (#43)
    "300404269": {"type": "classification", "label_en": "attributed to", "label_nl": "toegeschreven aan", "external_id": "http://vocab.getty.edu/aat/300404269"},
    "300404274": {"type": "classification", "label_en": "workshop of", "label_nl": "werkplaats van", "external_id": "http://vocab.getty.edu/aat/300404274"},
    "300404284": {"type": "classification", "label_en": "circle of", "label_nl": "omgeving van", "external_id": "http://vocab.getty.edu/aat/300404284"},
    "300404282": {"type": "classification", "label_en": "follower of", "label_nl": "navolger van", "external_id": "http://vocab.getty.edu/aat/300404282"},
    "300404272": {"type": "classification", "label_en": "manner of", "label_nl": "manier van", "external_id": "http://vocab.getty.edu/aat/300404272"},
    "300404279": {"type": "classification", "label_en": "copy after", "label_nl": "kopie naar", "external_id": "http://vocab.getty.edu/aat/300404279"},
    "300404434": {"type": "classification", "label_en": "school of", "label_nl": "school van", "external_id": "http://vocab.getty.edu/aat/300404434"},
    "300404273": {"type": "classification", "label_en": "studio of", "label_nl": "atelier van", "external_id": "http://vocab.getty.edu/aat/300404273"},
    # Additional attribution qualifiers — discovered via maker-relations probe (2026-03-08)
    "300404286": {"type": "classification", "label_en": "after", "label_nl": "naar", "external_id": "http://vocab.getty.edu/aat/300404286"},
    "300404287": {"type": "classification", "label_en": "copyist of", "label_nl": "imitator van", "external_id": "http://vocab.getty.edu/aat/300404287"},
    "300435722": {"type": "classification", "label_en": "possibly", "label_nl": "mogelijk", "external_id": "http://vocab.getty.edu/aat/300435722"},
    "300404283": {"type": "classification", "label_en": "circle of", "label_nl": "kring van", "external_id": "http://vocab.getty.edu/aat/300404283"},
    # Additional attribution qualifiers — discovered via v0.24 schema discovery
    "300404288": {"type": "classification", "label_en": "manner of", "label_nl": "op de manier van", "external_id": "http://vocab.getty.edu/aat/300404288"},
    "300252887": {"type": "classification", "label_en": "falsification", "label_nl": "vervalsing", "external_id": "http://vocab.getty.edu/aat/300252887"},
    # free-form: AAT form/shape attribute, not a maker-relation qualifier.
    # Seeded here because v0.24 found it referenced from attribution_qualifier
    # mappings — probable upstream classification quirk, see #227.
    "300312150": {"type": "classification", "label_en": "free-form", "label_nl": "free-form", "external_id": "http://vocab.getty.edu/aat/300312150"},
}

# Rijksmuseum vocab IDs confirmed 404 on id.rijksmuseum.nl. Excluded from
# the orphan audit so they stop re-triggering pre-harvest review. If an
# ID starts returning 200 on a future harvest, remove it from the set.
KNOWN_DEAD_URIS = {
    "2109224",    # 404 verified 2026-04-15
    "21053496",   # 404 verified 2026-04-15
    "21016167",   # 404 verified 2026-04-15
    "2306041",    # 404 verified 2026-04-15
}

# ─── XML Namespaces ──────────────────────────────────────────────────

NS = {
    "oai": "http://www.openarchives.org/OAI/2.0/",
    "dc": "http://purl.org/dc/elements/1.1/",
    "dcterms": "http://purl.org/dc/terms/",
    "edm": "http://www.europeana.eu/schemas/edm/",
    "edmfp": "http://www.europeanafashion.eu/edmfp/",
    "ore": "http://www.openarchives.org/ore/terms/",
    "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
    "rdaGr2": "http://rdvocab.info/ElementsGr2/",
    "skos": "http://www.w3.org/2004/02/skos/core#",
    "owl": "http://www.w3.org/2002/07/owl#",
}

RDF_RESOURCE = "{http://www.w3.org/1999/02/22-rdf-syntax-ns#}resource"
RDF_ABOUT = "{http://www.w3.org/1999/02/22-rdf-syntax-ns#}about"
OWL_SAME_AS = f"{{{NS['owl']}}}sameAs"
XML_LANG = "{http://www.w3.org/XML/1998/namespace}lang"

# Regex to extract IIIF identifier from iiif.micr.io URLs
IIIF_ID_RE = re.compile(r"https?://iiif\.micr\.io/([^/]+)")

# EDM fields on ProvidedCHO that map to vocabulary references.
# Each entry: (XML tag with full namespace, mapping field name)
CHO_VOCAB_FIELDS = [
    ("{http://purl.org/dc/elements/1.1/}subject",             "subject"),
    ("{http://purl.org/dc/terms/}medium",                     "material"),
    ("{http://purl.org/dc/elements/1.1/}type",                "type"),       # rdf:resource refs only, not text edm:type
    ("{http://purl.org/dc/elements/1.1/}creator",             "creator"),
    ("{http://purl.org/dc/terms/}spatial",                    "spatial"),     # production place
    ("{http://www.europeanafashion.eu/edmfp/}technique",      "technique"),
]


# ─── Database Schema ─────────────────────────────────────────────────

SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS vocabulary (
    id          TEXT PRIMARY KEY,
    type        TEXT NOT NULL,
    label_en    TEXT,
    label_nl    TEXT,
    external_id TEXT,
    broader_id  TEXT,
    notation    TEXT,
    lat             REAL,
    lon             REAL,
    label_en_norm   TEXT,
    label_nl_norm   TEXT
);

-- #238: Multi-valued external authority IDs (Wikidata, VIAF, RKD, ULAN, TGN, AAT, Iconclass, GeoNames, etc).
-- The legacy `vocabulary.external_id` column is kept as a single-pick hint (Wikidata preferred) for
-- backwards-compatible reads (src/api/VocabularyDb.ts), but this table holds the full set and is the
-- source of truth for reconciliation and authority-scoped lookups.
--
-- #268 audit-trail note: there is no separate `external_id_method` column on this table — the
-- `authority` column ('wikidata' / 'tgn' / 'geonames' / 'viaf' / 'rkd' / 'ulan' / 'aat' /
-- 'iconclass' / ...) IS the audit. Every row in this table came from a known external authority,
-- so the trust tier is implicitly 'deterministic'.
CREATE TABLE IF NOT EXISTS vocabulary_external_ids (
    vocab_id    TEXT NOT NULL,
    authority   TEXT NOT NULL,
    id          TEXT NOT NULL,
    uri         TEXT NOT NULL,
    PRIMARY KEY (vocab_id, authority, id)
) WITHOUT ROWID;
CREATE INDEX IF NOT EXISTS idx_vei_authority_id ON vocabulary_external_ids (authority, id);
CREATE INDEX IF NOT EXISTS idx_vei_uri          ON vocabulary_external_ids (uri);

CREATE TABLE IF NOT EXISTS artworks (
    object_number    TEXT PRIMARY KEY,
    title            TEXT,
    creator_label    TEXT,
    rights_uri       TEXT,
    linked_art_uri   TEXT,
    inscription_text TEXT,
    provenance_text  TEXT,
    credit_line      TEXT,
    description_text TEXT,
    height_cm        REAL,
    width_cm         REAL,
    narrative_text   TEXT,
    date_earliest    INTEGER,
    date_latest      INTEGER,
    title_all_text   TEXT,
    has_image        INTEGER DEFAULT 0,
    iiif_id          TEXT,
    tier2_done       INTEGER DEFAULT 0,
    date_display     TEXT,
    current_location TEXT,
    depth_cm         REAL,
    weight_g         REAL,
    diameter_cm      REAL,
    dimension_note   TEXT,
    provenance_text_hash TEXT,
    -- LDES record-edit timestamps (ISO 8601, nullable). Aggregated across
    -- referred_to_by[].created_by.timespan in the LA shape.
    record_created   TEXT,
    record_modified  TEXT,
    -- Free-text dimensions/extent string from dcterms:extent (CHO).
    extent_text      TEXT,
    -- HMO -> VisualItem cross-link from shows[0].id.
    visualitem_uri   TEXT
);

CREATE TABLE IF NOT EXISTS mappings (
    object_number  TEXT NOT NULL,
    vocab_id       TEXT NOT NULL,
    field          TEXT NOT NULL,
    PRIMARY KEY (object_number, vocab_id, field)
);

CREATE INDEX IF NOT EXISTS idx_vocab_label_en ON vocabulary(label_en COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_vocab_label_nl ON vocabulary(label_nl COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_vocab_notation ON vocabulary(notation);
CREATE INDEX IF NOT EXISTS idx_vocab_type ON vocabulary(type);
CREATE INDEX IF NOT EXISTS idx_artworks_tier2 ON artworks(tier2_done);

CREATE TABLE IF NOT EXISTS person_names (
    person_id       TEXT NOT NULL REFERENCES vocabulary(id),
    name            TEXT NOT NULL,
    lang            TEXT,
    classification  TEXT,
    UNIQUE(person_id, name, lang)
);
CREATE INDEX IF NOT EXISTS idx_person_names_id ON person_names(person_id);

CREATE TABLE IF NOT EXISTS exhibitions (
    exhibition_id INTEGER PRIMARY KEY,
    title_en      TEXT,
    title_nl      TEXT,
    date_start    TEXT,
    date_end      TEXT
);

CREATE TABLE IF NOT EXISTS exhibition_members (
    exhibition_id INTEGER NOT NULL,
    hmo_id        TEXT NOT NULL,
    PRIMARY KEY (exhibition_id, hmo_id)
);

CREATE TABLE IF NOT EXISTS modifications (
    art_id       INTEGER NOT NULL,
    seq          INTEGER NOT NULL,
    modifier_uri TEXT,
    date_display TEXT,
    date_begin   TEXT,
    date_end     TEXT,
    description  TEXT,
    PRIMARY KEY (art_id, seq)
) WITHOUT ROWID;

CREATE TABLE IF NOT EXISTS related_objects (
    art_id           INTEGER NOT NULL,
    related_la_uri   TEXT NOT NULL,
    related_art_id   INTEGER,
    relationship_en  TEXT NOT NULL,
    relationship_nl  TEXT,
    PRIMARY KEY (art_id, related_la_uri)
) WITHOUT ROWID;

CREATE TABLE IF NOT EXISTS examinations (
    art_id          INTEGER NOT NULL,
    seq             INTEGER NOT NULL,
    examiner_name   TEXT,
    report_type_id  TEXT NOT NULL,
    report_type_en  TEXT,
    date_display    TEXT,
    date_begin      TEXT,
    date_end        TEXT,
    PRIMARY KEY (art_id, seq)
) WITHOUT ROWID;

CREATE TABLE IF NOT EXISTS title_variants (
    art_id     INTEGER NOT NULL,
    seq        INTEGER NOT NULL,
    title_text TEXT NOT NULL,
    language   TEXT,
    qualifier  TEXT,
    PRIMARY KEY (art_id, seq)
) WITHOUT ROWID;

CREATE TABLE IF NOT EXISTS assignment_pairs (
    artwork_id   INTEGER NOT NULL,
    qualifier_id TEXT NOT NULL,
    creator_id   TEXT NOT NULL,
    part_index   INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (artwork_id, qualifier_id, creator_id)
) WITHOUT ROWID;
-- #144: qualifier-led aggregation index for findSimilarByLineage.
-- Covers (qualifier_id) for IDF, (qualifier_id, creator_id) for shared-pair
-- lookup. Without this, aggregation queries scan the PK in PK order.
CREATE INDEX IF NOT EXISTS idx_assignment_pairs_qualifier
  ON assignment_pairs(qualifier_id, creator_id, artwork_id);

CREATE TABLE IF NOT EXISTS artwork_parent (
    art_id        INTEGER NOT NULL,
    parent_la_uri TEXT NOT NULL,
    parent_art_id INTEGER,
    PRIMARY KEY (art_id, parent_la_uri)
) WITHOUT ROWID;
-- #267: reverse-edge index for parent → children lookups (per-detail
-- stmtArtworkChildCount/stmtArtworkChildrenPreview, plus groupBy=parent
-- in search_artwork). Forward art_id → parent is free via the PK prefix;
-- without this, parent_art_id queries SCAN ~208K rows.
CREATE INDEX IF NOT EXISTS idx_artwork_parent_reverse
  ON artwork_parent(parent_art_id);

CREATE TABLE IF NOT EXISTS phase2_failures (
    uri        TEXT PRIMARY KEY,
    reason     TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
CREATE INDEX IF NOT EXISTS idx_phase2_failures_reason ON phase2_failures(reason);

-- #226: Unified persistent failure log across phases 2, 2b, and 4.
-- Survives Phase 3 integer-encoding (which drops `tier2_done`), so post-harvest
-- diagnostics and targeted repair scripts have a stable query surface. Populated
-- after the Phase 2 retry sweep (#225) so only *persistent* failures land here —
-- transient ones already recovered on retry are removed.
CREATE TABLE IF NOT EXISTS phase_failures (
    phase         INTEGER NOT NULL,  -- 2, 4 (Phase 2b folds into 2)
    uri_or_objnum TEXT NOT NULL,
    failure_type  TEXT NOT NULL,     -- 'timeout', 'http_404', 'http_5xx', 'parse_error', 'unsupported_type:*', 'unknown'
    error_message TEXT,
    attempted_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    retry_count   INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (phase, uri_or_objnum)
) WITHOUT ROWID;
CREATE INDEX IF NOT EXISTS idx_phase_failures_type ON phase_failures(failure_type);

-- Attribution-evidence junction. Captures the
-- `produced_by.part[].assigned_by[].motivated_by[].carried_by[]` trail
-- as opaque rows: evidence type AAT URI, citation/publication URI,
-- free-text label.
CREATE TABLE IF NOT EXISTS attribution_evidence (
    art_id              INTEGER NOT NULL,
    part_index          INTEGER NOT NULL,
    evidence_type_aat   TEXT,
    carried_by_uri      TEXT,
    label_text          TEXT,
    PRIMARY KEY (art_id, part_index, evidence_type_aat, carried_by_uri)
) WITHOUT ROWID;
CREATE INDEX IF NOT EXISTS idx_attribution_evidence_art ON attribution_evidence(art_id);

-- Artwork-level external IDs (Wikidata, handle, etc. on the HMO itself,
-- distinct from `vocabulary_external_ids` which keys on vocab terms).
CREATE TABLE IF NOT EXISTS artwork_external_ids (
    art_id     INTEGER NOT NULL,
    authority  TEXT NOT NULL,
    id         TEXT NOT NULL,
    uri        TEXT NOT NULL,
    PRIMARY KEY (art_id, authority, id)
) WITHOUT ROWID;
CREATE INDEX IF NOT EXISTS idx_artwork_external_ids_authority ON artwork_external_ids(authority, id);

-- Polymorphic entity alt-names — covers entity types (organisations, etc.)
-- that don't have a dedicated alt-name table. person_names stays as-is for
-- the existing FTS surface.
-- NOTE: rowid-backed (not WITHOUT ROWID) because the FTS5 contentful twin
-- below needs the implicit rowid for content_rowid binding.
CREATE TABLE IF NOT EXISTS entity_alt_names (
    entity_id      TEXT NOT NULL,
    entity_type    TEXT NOT NULL,
    name           TEXT NOT NULL,
    lang           TEXT,
    classification TEXT,
    UNIQUE(entity_id, name)
);
CREATE INDEX IF NOT EXISTS idx_entity_alt_names_type ON entity_alt_names(entity_type);

CREATE VIRTUAL TABLE IF NOT EXISTS entity_alt_names_fts
USING fts5(name, content='entity_alt_names', content_rowid='rowid');
"""

VOCAB_INSERT_SQL = (
    "INSERT OR IGNORE INTO vocabulary "
    "(id, type, label_en, label_nl, external_id, broader_id, notation, lat, lon) "
    "VALUES (:id, :type, :label_en, :label_nl, :external_id, :broader_id, :notation, :lat, :lon)"
)

VEI_INSERT_SQL = (
    "INSERT OR IGNORE INTO vocabulary_external_ids "
    "(vocab_id, authority, id, uri) VALUES (?, ?, ?, ?)"
)


# Mappings INSERT shape diverges between text-schema (pre-Phase 3) and
# integer-schema (post-Phase 3 finalization). Phase 4 and Phase 4.5 both
# insert into mappings during the harvest, so they share this helper.
_MAPPING_INSERT_SQL_INT = """
    INSERT OR IGNORE INTO mappings (artwork_id, vocab_rowid, field_id)
    SELECT a.art_id, v.vocab_int_id, f.id
    FROM artworks a, vocabulary v, field_lookup f
    WHERE a.object_number = ? AND v.id = ? AND f.name = ?
"""

_MAPPING_INSERT_SQL_TEXT = (
    "INSERT OR IGNORE INTO mappings (object_number, vocab_id, field) VALUES (?, ?, ?)"
)


def _mapping_insert_sql(int_schema: bool) -> str:
    return _MAPPING_INSERT_SQL_INT if int_schema else _MAPPING_INSERT_SQL_TEXT


def get_columns(conn: sqlite3.Connection, table: str) -> set[str]:
    """Return the set of column names for a given table."""
    return {row[1] for row in conn.execute(f"PRAGMA table_info({table})").fetchall()}


def table_exists(conn: sqlite3.Connection, name: str) -> bool:
    """Check if a table exists in the database."""
    return conn.execute(
        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,)
    ).fetchone() is not None


def create_or_open_db() -> sqlite3.Connection:
    """Create or open the SQLite database, migrating schema if needed."""
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(DB_PATH))
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    conn.execute("PRAGMA cache_size=-64000")  # 64 MB cache
    conn.executescript(SCHEMA_SQL)

    # Create mappings indexes appropriate for the current schema.
    # Integer-encoded DBs (post-normalize_mappings) have field_id/artwork_id/vocab_rowid;
    # text DBs (fresh harvest, pre-Phase 3) have field/object_number/vocab_id.
    mapping_cols = get_columns(conn, "mappings")
    if "field_id" in mapping_cols:
        # Integer-encoded schema — indexes are created by normalize_mappings() in Phase 3.
        # Create them here too for idempotency (IF NOT EXISTS avoids duplicates).
        conn.execute("CREATE INDEX IF NOT EXISTS idx_mappings_field_vocab   ON mappings(field_id, vocab_rowid)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_mappings_field_artwork ON mappings(field_id, artwork_id)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_mappings_vocab         ON mappings(vocab_rowid)")
        # Drop stub index left over from v0.15 harvest workaround (duplicate of idx_mappings_field_artwork)
        conn.execute("DROP INDEX IF EXISTS idx_mappings_field_object")
    else:
        # Text schema — original indexes for Phase 1/2/4 queries
        conn.execute("CREATE INDEX IF NOT EXISTS idx_mappings_field_vocab  ON mappings(field, vocab_id)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_mappings_field_object ON mappings(field, object_number)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_mappings_vocab        ON mappings(vocab_id)")

    # Migrate existing DBs: add normalized label columns if missing
    vocab_cols = get_columns(conn, "vocabulary")
    for col_name, col_type in [("label_en_norm", "TEXT"), ("label_nl_norm", "TEXT")]:
        if col_name not in vocab_cols:
            conn.execute(f"ALTER TABLE vocabulary ADD COLUMN {col_name} {col_type}")
            print(f"  Migrated: added vocabulary.{col_name}")

    # Migrate existing DBs: add Tier 2 columns if missing
    existing_cols = get_columns(conn, "artworks")
    tier2_cols = [
        ("linked_art_uri", "TEXT"),
        ("inscription_text", "TEXT"),
        ("provenance_text", "TEXT"),
        ("credit_line", "TEXT"),
        ("description_text", "TEXT"),
        ("height_cm", "REAL"),
        ("width_cm", "REAL"),
        ("narrative_text", "TEXT"),
        ("date_earliest", "INTEGER"),
        ("date_latest", "INTEGER"),
        ("title_all_text", "TEXT"),
        ("tier2_done", "INTEGER DEFAULT 0"),
        ("date_display", "TEXT"),
        ("current_location", "TEXT"),
        ("depth_cm", "REAL"),
        ("weight_g", "REAL"),
        ("diameter_cm", "REAL"),
        ("dimension_note", "TEXT"),
        ("provenance_text_hash", "TEXT"),
        ("record_created", "TEXT"),
        ("record_modified", "TEXT"),
        ("extent_text", "TEXT"),
        ("visualitem_uri", "TEXT"),
    ]
    for col_name, col_type in tier2_cols:
        if col_name not in existing_cols:
            conn.execute(f"ALTER TABLE artworks ADD COLUMN {col_name} {col_type}")
            print(f"  Migrated: added artworks.{col_name}")
    conn.commit()

    return conn


# ─── Phase 0: Parse data dumps ────────────────────────────────────────


# Type alias: a callable taking an Iconclass notation string and returning a
# (label_en, label_nl) tuple if either is resolvable, else None. Built once
# per harvest by `make_iconclass_resolver()`. Used as a *last-resort* label
# fallback in `parse_nt_file` for classification entities whose dump file
# carries only a P190 notation string and no label literal of any kind.
IconclassResolver = "callable[[str], tuple[str | None, str | None] | None] | None"


def make_iconclass_resolver(db_path: str | None = None):
    """Return a label resolver backed by `iconclass.db` (the local sibling
    project's bundled Iconclass dump).

    Recovery for classification entities like `41D224(SIERMOUWEN)` whose
    dump file carries the notation but no label literal (#245). Three-step
    lookup, first match wins:

      1. **exact** — the full notation as-is
      2. **paren-stripped** — drop Rijksmuseum-specific keys like
         `(SIERMOUWEN)`, `(TURTLE)` to recover the canonical Iconclass base
         (`41D224(SIERMOUWEN)` → `41D224` → "sleeves"). The Rijksmuseum
         extends Iconclass with Dutch keyword keys; iconclass.org's dump
         only has the canonical notations.
      3. **progressive prefix** — shave one trailing char at a time until
         a match (`41D2619(+82)` → `41D2619` → `41D261` → `41D26` → ...).
         Always semantically valid: every prefix of an Iconclass notation
         is a broader category. Recovery degrades gracefully (less specific
         label) but never lies.

    Returns None if the resolver is unavailable (no DB path, file missing).
    Caller threads the result through `parse_nt_file(..., iconclass_resolver=)`.

    DB path resolution order: explicit arg > `ICONCLASS_DB_PATH` env var >
    project-local `data/iconclass.db` > sibling-checkout
    `../rijksmuseum-iconclass-mcp/data/iconclass.db` (the dev convention —
    the iconclass MCP server bundles its own 3.4 GB Iconclass dump).
    """
    if not db_path:
        db_path = os.environ.get("ICONCLASS_DB_PATH")
    if not db_path:
        project_root = Path(__file__).resolve().parent.parent
        candidates = [
            project_root / "data" / "iconclass.db",
            project_root.parent / "rijksmuseum-iconclass-mcp" / "data" / "iconclass.db",
        ]
        for c in candidates:
            if c.exists():
                db_path = str(c)
                break
    if not db_path or not os.path.exists(db_path):
        return None

    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
    paren_re = re.compile(r'\([^)]+\)')

    def resolve(notation: str) -> "tuple[str | None, str | None] | None":
        if not notation:
            return None
        # Build candidate list in resolution-order priority
        candidates: list[str] = [notation]
        base = paren_re.sub("", notation)
        if base != notation and base:
            candidates.append(base)
        target = base if base else notation
        while len(target) > 1:
            target = target[:-1]
            candidates.append(target)
        # Try each; first hit wins. Pull both en/nl in one query.
        for cand in candidates:
            rows = conn.execute(
                "SELECT lang, text FROM texts WHERE notation = ? AND lang IN ('en', 'nl')",
                (cand,),
            ).fetchall()
            if not rows:
                continue
            en = next((t for l, t in rows if l == "en"), None)
            nl = next((t for l, t in rows if l == "nl"), None)
            if en or nl:
                return (en, nl)
        return None

    return resolve


def parse_nt_file(filepath: str, default_type: str, iconclass_resolver=None,
                  parse_stats: Counter | None = None) -> dict | None:
    """Parse a single N-Triples entity file into a vocabulary record.

    Handles two coexisting shapes in the dumps:

    1. Linked Art / CIDOC-CRM — blank-node appellation chains
       (`E33_E41_Linguistic_Appellation` → `P190_has_symbolic_content`
       with `P72_has_language` pointing at an AAT language URI).
       Used by classification, concept, event dumps and ~88% of place.

    2. Schema.org — direct-on-subject `schema:name` literal + optional
       `schema:alternateName` + `schema:sameAs` authority links, with
       `rdf:type schema:{Person|Organization|DefinedTerm|Place}`.
       Used by person, organisation, topical_term dumps and ~12% of
       place (the hybrid place subset with TGN/Wikidata cross-refs).

    Returns a dict ready for VOCAB_INSERT_SQL binding, plus a special
    `_external_ids` key (leading underscore — not for SQL bind; the
    caller pops it before executemany) carrying a list of
    (authority, local_id, uri) tuples for vocabulary_external_ids.
    """
    entity_id = os.path.basename(filepath)
    entity_uri = f"https://id.rijksmuseum.nl/{entity_id}"

    bnodes: dict[str, dict] = {}
    equivalents: list[str] = []
    broader_id: str | None = None
    notation: str | None = None
    defined_by: str | None = None
    rdf_type: str | None = None
    skos_label_en: str | None = None
    skos_label_nl: str | None = None
    # Other-language label fallbacks (#245): used only when en/nl labels
    # are completely absent. Captures non-en/non-nl rdfs:label / skos:prefLabel
    # literals (first-wins) so we don't silently drop entities whose only
    # labels are French, German, etc.
    skos_label_other: str | None = None
    skos_label_untagged: str | None = None

    # Schema.org shape collectors (#238)
    schema_types: set[str] = set()
    schema_names_bare: list[str] = []           # literals with no language tag
    schema_names_by_lang: dict[str, str] = {}   # primary-subtag → text (first wins)
    schema_alt_names_bare: list[str] = []
    schema_alt_names_by_lang: dict[str, str] = {}
    schema_same_as: list[str] = []

    try:
        with open(filepath, "r", encoding="utf-8") as f:
            # #317: upstream dump-build occasionally saves an HTML error page (e.g.
            # a 404) under a .nt filename. It is not RDF — count it as a distinct
            # failure category instead of letting it pass as a silent zero-triple
            # drop. Real N-Triples lines start with <http...> or _:bnode (never
            # <!doctype / <html), so this cannot false-positive on valid content.
            head = f.read(64).lstrip().lower()
            if head.startswith(("<!doctype", "<html")):
                if parse_stats is not None:
                    parse_stats["html_error"] += 1
                return None
            f.seek(0)
            lines = f.readlines()
    except Exception:
        return None

    for line in lines:
        line = line.strip()
        if not line:
            continue

        m = NT_PATTERN.match(line)
        if m and m.group(1) == entity_uri:
            pred = m.group(2)
            obj_uri = m.group(3)
            obj_lit = _unescape_nt_literal(m.group(4)) if m.group(4) is not None else None
            if pred == P_EQUIVALENT and obj_uri:
                equivalents.append(obj_uri)
            elif pred == P_BROADER and obj_uri:
                broader_id = obj_uri.split("/")[-1]
            elif pred == "http://www.cidoc-crm.org/cidoc-crm/P168_place_is_defined_by" and obj_lit:
                defined_by = obj_lit
            elif pred == RDF_TYPE and obj_uri:
                rdf_type = obj_uri
                if obj_uri in SCHEMA_TYPE_MAP:
                    schema_types.add(obj_uri)
            # Schema.org: bare-literal variants (no language tag) match NT_PATTERN's
            # literal branch because there's no `@lang` suffix. Person/Organization
            # names come through here. Topical_term / place labels are language-tagged
            # and go through NT_LANG_PATTERN below instead.
            elif pred == P_SCHEMA_NAME and obj_lit is not None:
                schema_names_bare.append(obj_lit)
            elif pred == P_SCHEMA_ALTERNATE_NAME and obj_lit is not None:
                schema_alt_names_bare.append(obj_lit)
            elif pred == P_SCHEMA_SAME_AS and obj_uri:
                schema_same_as.append(obj_uri)
            elif pred in (SKOS_PREFLABEL, RDFS_LABEL) and obj_lit is not None:
                # #245 fallback: untagged rdfs:label / skos:prefLabel literal
                # (no `@lang` suffix). NT_LANG_PATTERN doesn't match these,
                # so without this branch they'd be silently dropped.
                if skos_label_untagged is None:
                    skos_label_untagged = obj_lit

        m = BNODE_PATTERN.match(line)
        if m:
            bnode_id = m.group(1)
            pred = m.group(2)
            obj_uri = m.group(3)
            obj_lit = _unescape_nt_literal(m.group(4)) if m.group(4) is not None else None

            if bnode_id not in bnodes:
                bnodes[bnode_id] = {}

            if pred == P_LABEL and obj_lit is not None:
                bnodes[bnode_id]["label"] = obj_lit
            elif pred == P_LANGUAGE and obj_uri:
                bnodes[bnode_id]["language"] = obj_uri
            elif pred == P_HAS_TYPE and obj_uri == AAT_DISPLAY_NAME:
                bnodes[bnode_id]["is_display_name"] = True
            elif pred == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type":
                if obj_uri == "http://www.cidoc-crm.org/cidoc-crm/E42_Identifier":
                    bnodes[bnode_id]["is_identifier"] = True

        # Match language-tagged literals (skos:prefLabel, rdfs:label, schema:name)
        m = NT_LANG_PATTERN.match(line)
        if m and m.group(1) == entity_uri:
            pred = m.group(2)
            text = _unescape_nt_literal(m.group(3))
            lang_tag = m.group(4)
            if pred in (SKOS_PREFLABEL, RDFS_LABEL) and text:
                if lang_tag == "en" and not skos_label_en:
                    skos_label_en = text
                elif lang_tag == "nl" and not skos_label_nl:
                    skos_label_nl = text
                elif skos_label_other is None:
                    # #245 fallback: capture first non-en/non-nl tag as a
                    # last resort. lang_tag is always present here because
                    # NT_LANG_PATTERN requires the @lang suffix.
                    skos_label_other = text
            elif pred == P_SCHEMA_NAME and text:
                # BCP 47 primary subtag (e.g. `nl-NL` → `nl`, `zh-Hans-CN` → `zh`)
                primary = lang_tag.split("-", 1)[0].lower()
                if primary not in schema_names_by_lang:
                    schema_names_by_lang[primary] = text
            elif pred == P_SCHEMA_ALTERNATE_NAME and text:
                primary = lang_tag.split("-", 1)[0].lower()
                if primary not in schema_alt_names_by_lang:
                    schema_alt_names_by_lang[primary] = text

    label_en = None
    label_nl = None
    bnode_label_other: str | None = None  # #245 fallback for non-en/non-nl display-name labels

    for bdata in bnodes.values():
        label = bdata.get("label")
        if not label:
            continue
        if bdata.get("is_identifier"):
            notation = label
        elif bdata.get("is_display_name"):
            lang = bdata.get("language")
            if lang == LANG_EN:
                label_en = label
            elif lang == LANG_NL:
                label_nl = label
            elif bnode_label_other is None:
                # Non-en/non-nl display-name label (or no language URI): keep
                # as last-resort fallback. First-wins; ordering across bnodes
                # is dict-iteration order which is dump-file dependent.
                bnode_label_other = label

    # Fallback: use skos/rdfs labels if display-name labels weren't found
    if not label_en:
        label_en = skos_label_en
    if not label_nl:
        label_nl = skos_label_nl

    # Determine type from RDF type if we can
    vocab_type = default_type
    if rdf_type:
        type_name = rdf_type.rsplit("/", 1)[-1] if "/" in rdf_type else rdf_type
        type_name = type_name.rsplit("#", 1)[-1] if "#" in type_name else type_name
        if type_name in LA_TYPE_MAP:
            vocab_type = LA_TYPE_MAP[type_name]

    # Schema.org branch: if the entity declares a schema:* type, overlay the
    # Schema.org-derived labels/type/sameAs on top of anything picked up
    # above. Schema.org and Linked Art don't co-occur in the same file
    # (verified via full-sweep: 0 entities have both a direct schema:name
    # and a blank-node appellation), but place files can carry both
    # `rdf:type schema:Place` AND `rdf:type E53_Place`, so the
    # type-precedence rule is: Schema.org type wins when present.
    if schema_types:
        for st in schema_types:
            vocab_type = SCHEMA_TYPE_MAP[st]
            break  # files only ever have one schema type in practice

        # Label selection: prefer language-tagged (nl first, then en),
        # then bare literals, then alternateName as a last resort (covers
        # the 8 anonymous entities — 7 persons + 1 organisation — whose
        # `schema:name` is absent entirely, #238 full sweep).
        if not label_nl:
            label_nl = schema_names_by_lang.get("nl")
        if not label_en:
            label_en = schema_names_by_lang.get("en")
        if not label_nl and not label_en and schema_names_bare:
            # Bare literal: assign as label_en by convention (persons/orgs
            # carry no language tag and labels like "Ahrendts" are
            # effectively language-agnostic).
            label_en = schema_names_bare[0]
        if not label_nl and not label_en:
            # Last-resort alternateName fallback for anonymous entities
            if schema_alt_names_by_lang.get("nl"):
                label_nl = schema_alt_names_by_lang["nl"]
            elif schema_alt_names_by_lang.get("en"):
                label_en = schema_alt_names_by_lang["en"]
            elif schema_alt_names_bare:
                label_en = schema_alt_names_bare[0]

    # Merge `schema:sameAs` URIs with Linked Art `linked.art/equivalent`
    # URIs into a single ordered list for authority classification + the
    # legacy `external_id` column pick. `dict.fromkeys` preserves insertion
    # order (Python ≥3.7) while deduping in O(n).
    all_external_uris: list[str] = list(dict.fromkeys(equivalents + schema_same_as))

    # Pick best external ID (prefer Wikidata, Iconclass) — legacy column,
    # used as a hint by src/api/VocabularyDb.ts toWikidataUri()
    external_id = None
    for eq in all_external_uris:
        if "iconclass.org" in eq:
            external_id = eq
            break
        if "wikidata.org" in eq:
            external_id = eq
            break
    if not external_id and all_external_uris:
        external_id = all_external_uris[0]

    # Classify all URIs for the multi-valued vocabulary_external_ids table.
    # Deduplicate on (authority, local_id) — a single entity sometimes
    # carries two URIs that normalise to the same authority-local pair.
    external_ids_out: list[tuple[str, str, str]] = []
    seen: set[tuple[str, str]] = set()
    for uri in all_external_uris:
        authority, local_id = classify_authority(uri)
        key = (authority, local_id)
        if key in seen:
            continue
        seen.add(key)
        external_ids_out.append((authority, local_id, uri))

    # Parse coordinates for places. #328: do NOT assign the WKT POINT to
    # `notation` — that column is reserved for Iconclass classifications.
    lat = None
    lon = None
    if defined_by and defined_by.startswith("POINT"):
        m_coord = re.match(r"POINT\(([-\d.]+)\s+([-\d.]+)\)", defined_by)
        if m_coord:
            lon = float(m_coord.group(1))
            lat = float(m_coord.group(2))

    # #245: Other-language fallback before final drop. Recovers entities
    # whose only labels are in French/German/etc. or carry no @lang tag.
    # Stored under label_en by convention (the column is treated as the
    # primary display label downstream; precise BCP47 tagging is out of
    # scope for the vocab table). Tried in priority order:
    #   1. blank-node display-name with non-en/non-nl AAT language URI
    #   2. skos:prefLabel / rdfs:label with non-en/non-nl @lang
    #   3. skos:prefLabel / rdfs:label with no @lang at all
    #   4. any schema:name in any language (covers files that have only
    #      a single non-en/non-nl @lang-tagged schema:name and no other
    #      label channels)
    #   5. any schema:alternateName in any language
    if not label_en and not label_nl:
        fallback = (
            bnode_label_other
            or skos_label_other
            or skos_label_untagged
            or next(iter(schema_names_by_lang.values()), None)
            or next(iter(schema_alt_names_by_lang.values()), None)
        )
        if fallback:
            label_en = fallback
        elif vocab_type == "event":
            label_en = entity_id  # Use entity ID as fallback label
        elif iconclass_resolver is not None and vocab_type == "classification":
            # #245 Tier 1: last-resort lookup against iconclass.db. Two ways
            # to derive a candidate notation for a label-less classification:
            #   1. P190 literal on the E42_Identifier bnode (most files)
            #   2. URL path of any iconclass.org URI in `linked.art/equivalent`
            #      (the ~11 files that carry only the URI, no P190 literal —
            #      e.g. `<...> equivalent <https://iconclass.org/61BB11%28%2B0%29>`)
            # Places never reach this branch (vocab_type == "classification"
            # required above), so `notation` here is always either a real
            # Iconclass code or None.
            candidate_notation: str | None = None
            if notation:
                candidate_notation = notation
            else:
                for eq in equivalents:
                    if "iconclass.org/" in eq:
                        candidate_notation = urllib.parse.unquote(
                            eq.split("iconclass.org/", 1)[1]
                        )
                        if not notation:
                            notation = candidate_notation  # also persist for audit
                        break
            if candidate_notation:
                resolved = iconclass_resolver(candidate_notation)
                if resolved:
                    label_en, label_nl = resolved
                else:
                    return None
            else:
                return None
        else:
            return None

    return {
        "id": entity_id,
        "type": vocab_type,
        "label_en": label_en,
        "label_nl": label_nl,
        "external_id": external_id,
        "broader_id": broader_id,
        "notation": notation,
        "lat": lat,
        "lon": lon,
        "_external_ids": external_ids_out,  # popped by run_phase0 before SQL bind
    }


def extract_dump(tar_name: str) -> Path | None:
    """Extract a tar.gz dump to a temp directory, return the path."""
    tar_path = DUMPS_DIR / f"{tar_name}.tar.gz"
    if not tar_path.exists():
        return None
    extract_dir = Path("/tmp") / f"rm-dump-{tar_name}"
    if extract_dir.exists() and any(extract_dir.iterdir()):
        return extract_dir  # Already extracted
    extract_dir.mkdir(parents=True, exist_ok=True)
    with tarfile.open(tar_path, "r:gz") as tf:
        tf.extractall(extract_dir, filter="data")
    return extract_dir


def parse_dump_dir(dump_dir: Path, default_type: str, iconclass_resolver=None) -> list[dict]:
    """Parse all N-Triples files in a dump directory."""
    files = [f for f in os.listdir(dump_dir) if os.path.isfile(dump_dir / f) and not f.startswith(".")]
    total = len(files)
    records = []
    parse_stats: Counter = Counter()
    for i, fname in enumerate(files):
        if i % 5000 == 0 and i > 0:
            print(f"    Parsing: {i}/{total}...", flush=True)
        rec = parse_nt_file(str(dump_dir / fname), default_type,
                            iconclass_resolver=iconclass_resolver, parse_stats=parse_stats)
        if rec:
            records.append(rec)
    if parse_stats["html_error"]:
        # #317: surface upstream HTML-instead-of-RDF as a counted category rather
        # than a silent zero-triple drop. Re-run scripts/probe_317_html_dumps.py
        # after a dump refresh to confirm which IDs are affected.
        print(f"    WARNING: {parse_stats['html_error']} file(s) in {dump_dir.name} were "
              f"HTML error pages, not RDF (#317) — upstream dump-build saved error "
              f"responses under .nt filenames; skipped.", flush=True)
    return records


def _strip_time(iso_literal: str) -> str:
    """Normalize an ISO datetime literal to YYYY-MM-DD.

    P82a_begin_of_the_begin / P82b_end_of_the_end values in the exhibition
    dump are typically '2006-08-11T00:00:00Z'. Consumers of the exhibitions
    table want a plain date string, so we strip the time portion.
    """
    return iso_literal.split("T")[0]


def parse_exhibition_dump(dump_dir: Path, conn: sqlite3.Connection) -> tuple[int, int]:
    """Parse exhibition dump files and populate exhibitions + exhibition_members tables.

    Exhibition entities are E7_Activity with ID namespace 241xxxx. The dump uses
    three parallel blank-node chains from the exhibition entity — the parser
    must follow all three:

    1. Titles via P1_is_identified_by chain:
         <Exhibition> P1_is_identified_by _:titleBnode
         _:titleBnode  P190_has_symbolic_content "real title"
       Title bnodes typically carry NO P72_has_language tag, so the real
       disambiguator is reachability from the exhibition entity via
       P1_is_identified_by — NOT language-based filtering. (The timespan's
       nested display-name bnodes DO carry language tags with date-range
       strings like "2006-08-11 - 2006-10-11" which historically got wrongly
       picked up as titles.)

    2. Dates via P4_has_time-span chain:
         <Exhibition> P4_has_time-span _:timespanBnode
         _:timespanBnode P82a_begin_of_the_begin "2006-08-11T00:00:00Z"
         _:timespanBnode P82b_end_of_the_end     "2006-10-11T23:59:59Z"
       Dates are on the timespan bnode, not on the exhibition entity itself.
       The legacy direct-entity path (NT_PATTERN P_BEGIN/P_END) is retained
       as a defensive fallback but does not trigger on current dumps.

    3. Members via P16_used_specific_object → has_member chain (#220):
         <Exhibition> P16_used_specific_object _:Set
         _:Set        https://linked.art/ns/terms/has_member <HMO>
       Legacy CIDOC-CRM direct/inverse membership predicates
       (P46_is_composed_of, P46i_forms_part_of) are retained as defensive
       fallbacks but are not used by the current dump.

    See issues #220 (has_member) and #236 (titles/dates) for the diagnosis.

    Returns (exhibition_count, member_count).
    """
    files = [f for f in os.listdir(dump_dir) if os.path.isfile(dump_dir / f) and not f.startswith(".")]
    exhibition_count = 0
    member_count = 0

    for fname in files:
        entity_id = fname
        entity_uri = f"https://id.rijksmuseum.nl/{entity_id}"
        filepath = dump_dir / fname

        bnodes: dict[str, dict] = {}
        date_begin = None
        date_end = None
        member_uris: set[str] = set()
        # Blank nodes referenced by this exhibition via P16_used_specific_object.
        # These are the Set entities whose has_member edges give us members.
        set_bnodes: set[str] = set()
        # Blank nodes referenced by this exhibition via P1_is_identified_by.
        # These are the title entities whose P190 labels give us real titles.
        # Distinct from the timespan's nested display-name bnodes, which would
        # otherwise wrongly win a language-based title race.
        title_bnodes: set[str] = set()
        # Blank nodes referenced by this exhibition via P4_has_time-span.
        # These are the timespan entities whose P82a/P82b literals give us dates.
        timespan_bnodes: set[str] = set()
        # Buffer: all bnode → HMO has_member edges seen in this file, keyed by bnode.
        # Filtered against set_bnodes after the parse loop to keep only the ones
        # that belong to this exhibition's Set(s).
        bnode_members: dict[str, set[str]] = {}
        # Buffer: all bnode → begin/end date literals seen in this file, keyed
        # by bnode. Filtered against timespan_bnodes after the parse loop.
        bnode_dates: dict[str, dict[str, str]] = {}

        try:
            with open(filepath, "r", encoding="utf-8") as f:
                lines = f.readlines()
        except Exception:
            continue

        for line in lines:
            line = line.strip()
            if not line:
                continue

            m = NT_PATTERN.match(line)
            if m:
                subj = m.group(1)
                pred = m.group(2)
                obj_uri = m.group(3)
                obj_lit = m.group(4)

                if subj == entity_uri:
                    if pred == P_BEGIN and obj_lit:
                        date_begin = obj_lit
                    elif pred == P_END and obj_lit:
                        date_end = obj_lit
                    # Legacy direct membership — retained as defensive fallback
                    elif pred == P_COMPOSED_OF and obj_uri:
                        hmo_num = obj_uri.split("/")[-1]
                        if hmo_num.isdigit():
                            member_uris.add(hmo_num)

                # Legacy inverse membership — retained as defensive fallback
                if obj_uri == entity_uri and pred == P_HAS_MEMBER:
                    hmo_num = subj.split("/")[-1]
                    if hmo_num.isdigit():
                        member_uris.add(hmo_num)
                continue

            mb = NT_TO_BNODE_PATTERN.match(line)
            if mb:
                subj = mb.group(1)
                pred = mb.group(2)
                bnode_id = mb.group(3)
                if subj == entity_uri:
                    if pred == P_USED_SPECIFIC_OBJECT:
                        set_bnodes.add(bnode_id)
                    elif pred == P_IS_IDENTIFIED_BY:
                        title_bnodes.add(bnode_id)
                    elif pred == P_HAS_TIME_SPAN:
                        timespan_bnodes.add(bnode_id)
                continue

            bm = BNODE_PATTERN.match(line)
            if bm:
                bnode_id = bm.group(1)
                pred = bm.group(2)
                obj_uri = bm.group(3)
                obj_lit = bm.group(4)

                if bnode_id not in bnodes:
                    bnodes[bnode_id] = {}

                if pred == P_LABEL and obj_lit is not None:
                    bnodes[bnode_id]["label"] = obj_lit
                elif pred == P_LANGUAGE and obj_uri:
                    bnodes[bnode_id]["language"] = obj_uri
                elif pred == P_HAS_MEMBER_LA and obj_uri:
                    # Buffer — filter against set_bnodes after the loop, since
                    # the triple order in the N-Triples file is not guaranteed
                    # and the P16 link may come after its has_member edges.
                    hmo_num = obj_uri.split("/")[-1]
                    if hmo_num.isdigit():
                        bnode_members.setdefault(bnode_id, set()).add(hmo_num)
                elif pred == P_BEGIN and obj_lit:
                    # Buffer — filter against timespan_bnodes after the loop.
                    # Triple order is not guaranteed; the P4 link may appear
                    # after the P82a/P82b literals in the file.
                    bnode_dates.setdefault(bnode_id, {})["begin"] = obj_lit
                elif pred == P_END and obj_lit:
                    bnode_dates.setdefault(bnode_id, {})["end"] = obj_lit

        # Resolve has_member edges: keep only those belonging to this exhibition's Set(s)
        for bnode_id in set_bnodes:
            member_uris.update(bnode_members.get(bnode_id, set()))

        # Resolve dates: filter buffered P82a/P82b to this exhibition's timespan bnodes.
        # date_begin / date_end may already be set from the legacy NT_PATTERN direct-
        # entity path above; the timespan-bnode path is the real one for current dumps.
        for ts_bnode in timespan_bnodes:
            d = bnode_dates.get(ts_bnode, {})
            if "begin" in d and date_begin is None:
                date_begin = _strip_time(d["begin"])
            if "end" in d and date_end is None:
                date_end = _strip_time(d["end"])

        # Resolve titles: filter labels to this exhibition's title bnodes. Title bnodes
        # typically have no P72_has_language tag (the real disambiguator is
        # P1_is_identified_by reachability, not language). Sample data shows exactly
        # one title bnode per exhibition, so we take the first non-empty label and
        # write it to both title_en and title_nl — the language is indeterminate and
        # the Rijksmuseum stores a single canonical title per exhibition.
        #
        # LATENT LIMITATION: if a future dump emits bilingual title variants on separate
        # bnodes (two bnodes reachable via P1_is_identified_by with different language
        # tags), this loop's `break` would take only the first. Refine at that point.
        title_text = None
        for t_bnode in title_bnodes:
            label = bnodes.get(t_bnode, {}).get("label")
            if label:
                title_text = label
                break

        title_en = title_text
        title_nl = title_text

        if not title_en and not title_nl:
            continue

        try:
            exh_id = int(entity_id)
        except ValueError:
            continue

        conn.execute(
            "INSERT OR IGNORE INTO exhibitions (exhibition_id, title_en, title_nl, date_start, date_end) VALUES (?, ?, ?, ?, ?)",
            (exh_id, title_en, title_nl, date_begin, date_end),
        )
        exhibition_count += 1

        for hmo_id in member_uris:
            conn.execute(
                "INSERT OR IGNORE INTO exhibition_members (exhibition_id, hmo_id) VALUES (?, ?)",
                (exh_id, hmo_id),
            )
            member_count += 1

    conn.commit()
    return exhibition_count, member_count


def _flush_vocab_batch(conn: sqlite3.Connection, records: list[dict]) -> int:
    """Insert a batch of parsed records into vocabulary + vocabulary_external_ids.

    Mutates `records` in place by popping `_external_ids` off each dict so
    the remaining keys match VOCAB_INSERT_SQL's named bindings. Returns
    the number of external-ID rows inserted.
    """
    ext_rows: list[tuple[str, str, str, str]] = []
    for r in records:
        vocab_id = r["id"]
        for authority, local_id, uri in r.pop("_external_ids"):
            ext_rows.append((vocab_id, authority, local_id, uri))
    conn.executemany(VOCAB_INSERT_SQL, records)
    if ext_rows:
        conn.executemany(VEI_INSERT_SQL, ext_rows)
    return len(ext_rows)


def run_phase0(conn: sqlite3.Connection):
    """Phase 0: Parse all data dumps into vocabulary table."""
    # Always seed external vocabulary entries regardless of dump availability
    for ext_id, ext_data in EXTERNAL_VOCAB.items():
        conn.execute(VOCAB_INSERT_SQL, {
            "id": ext_id, "type": ext_data["type"],
            "label_en": ext_data["label_en"], "label_nl": ext_data["label_nl"],
            "external_id": ext_data["external_id"],
            "broader_id": None, "notation": None, "lat": None, "lon": None,
        })
        # Also seed into vocabulary_external_ids so the new table is
        # consistent with the legacy column for these hand-curated entries.
        if ext_data.get("external_id"):
            authority, local_id = classify_authority(ext_data["external_id"])
            conn.execute(VEI_INSERT_SQL, (ext_id, authority, local_id, ext_data["external_id"]))
    conn.commit()
    print(f"  Seeded {len(EXTERNAL_VOCAB)} external vocabulary entries (Getty AAT)")

    if not DUMPS_DIR.is_dir():
        print(f"  SKIP: Data dumps not found at {DUMPS_DIR}")
        print(f"  Download from: https://data.rijksmuseum.nl/object-metadata/download/")
        return

    # #245 Tier 1: build the iconclass-fallback resolver once. None if the
    # sibling iconclass.db isn't available — harvest still works, just with
    # ~100 fewer recovered classifications.
    iconclass_resolver = make_iconclass_resolver()
    if iconclass_resolver is None:
        print("  Note: iconclass.db not found — classification label fallback disabled (#245).")
    else:
        print("  Iconclass-fallback resolver active for label-less classifications (#245).")

    total_records = 0
    total_ext_ids = 0

    for tar_name, default_type in DUMP_CONFIGS:
        print(f"  Processing {tar_name} dump (type={default_type})...")
        dump_dir = extract_dump(tar_name)
        if dump_dir is None:
            print(f"    SKIP: {tar_name}.tar.gz not found")
            continue

        # Exhibition dump gets special parsing (structured tables, not just vocab)
        if tar_name == "exhibition":
            exh_count, mem_count = parse_exhibition_dump(dump_dir, conn)
            # Also parse as regular vocab (for label lookup)
            records = parse_dump_dir(dump_dir, default_type, iconclass_resolver=iconclass_resolver)
            if records:
                ext_count = _flush_vocab_batch(conn, records)
                conn.commit()
                total_records += len(records)
                total_ext_ids += ext_count
            print(f"    {exh_count:,} exhibitions, {mem_count:,} artwork memberships, {len(records):,} vocab records")
            continue

        records = parse_dump_dir(dump_dir, default_type, iconclass_resolver=iconclass_resolver)
        if not records:
            print(f"    No records parsed")
            continue

        with_notation = sum(1 for r in records if r.get("notation"))
        ext_count = _flush_vocab_batch(conn, records)
        conn.commit()

        print(f"    {len(records):,} records ({with_notation:,} with notation, {ext_count:,} external IDs)")
        total_records += len(records)
        total_ext_ids += ext_count

    print(f"  Total: {total_records:,} vocabulary records, {total_ext_ids:,} external IDs from dumps")


# ─── Phase 0.5: Seed Set Names ───────────────────────────────────────

def run_phase0_5(conn: sqlite3.Connection):
    """Seed vocabulary table with curated set names from OAI-PMH ListSets."""
    url = f"{OAI_BASE}?verb=ListSets"
    print("  Fetching ListSets from OAI-PMH...")
    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with urllib.request.urlopen(req, timeout=60) as resp:
        root = ET.fromstring(resp.read())

    count = 0
    for set_el in root.findall(".//oai:set", NS):
        spec_el = set_el.find("oai:setSpec", NS)
        name_el = set_el.find("oai:setName", NS)
        if spec_el is None or name_el is None:
            continue
        set_spec = spec_el.text or ""
        set_name = name_el.text or ""
        if not set_spec or not set_name:
            continue
        conn.execute(VOCAB_INSERT_SQL, {
            "id": set_spec, "type": "set",
            "label_en": set_name, "label_nl": set_name,
            "external_id": None, "broader_id": None,
            "notation": None, "lat": None, "lon": None,
        })
        count += 1

    conn.commit()
    print(f"  Seeded {count} curated set names")


# ─── Phase 1: OAI-PMH Harvest ───────────────────────────────────────

def fetch_oai_page(url: str) -> ET.Element:
    """Fetch and parse an OAI-PMH XML page."""
    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    for attempt in range(3):
        try:
            with urllib.request.urlopen(req, timeout=60) as resp:
                return ET.fromstring(resp.read())
        except Exception as e:
            if attempt < 2:
                wait = 5 * (attempt + 1)
                print(f"  Retry {attempt + 1}/3 after error: {e} (waiting {wait}s)", flush=True)
                time.sleep(wait)
            else:
                raise


def extract_resource_ref(elem) -> str | None:
    """Extract vocabulary ID from an element.

    Handles two RDF/XML serialization patterns:
      1. Flat attribute:  <el rdf:resource="https://id.rijksmuseum.nl/12345"/>
      2. Nested element:  <el><edm:Place rdf:about="https://id.rijksmuseum.nl/12345">...
    """
    # Pattern 1: flat rdf:resource attribute
    ref = elem.get(RDF_RESOURCE, "")
    if ref:
        return ref.split("/")[-1]
    # Pattern 2: nested child element with rdf:about
    for child in elem:
        ref = child.get(RDF_ABOUT, "")
        if ref:
            return ref.split("/")[-1]
    return None


def extract_records(root: ET.Element) -> list[dict]:
    """Extract artwork metadata and vocabulary mappings from OAI-PMH EDM records."""
    records = []

    for record in root.findall(".//oai:record", NS):
        header = record.find("oai:header", NS)
        if header is None:
            continue
        # Skip deleted records
        if header.get("status") == "deleted":
            continue

        # Extract setSpec values from header (collection set memberships)
        set_specs = [el.text for el in header.findall("oai:setSpec", NS) if el.text]

        metadata = record.find("oai:metadata", NS)
        if metadata is None:
            continue

        # Find ProvidedCHO — nested inside rdf:RDF > ore:Aggregation > edm:aggregatedCHO
        cho = metadata.find(".//{http://www.europeana.eu/schemas/edm/}ProvidedCHO")
        if cho is None:
            continue

        # Extract Linked Art URI from CHO's rdf:about (for Phase 4 resolution)
        lod_uri = cho.get(RDF_ABOUT, "")

        # Extract object number
        object_number = ""
        id_el = cho.find("{http://purl.org/dc/elements/1.1/}identifier")
        if id_el is not None and id_el.text:
            object_number = id_el.text.strip()

        if not object_number:
            continue

        # Extract title (prefer English)
        title = ""
        for t in cho.findall("{http://purl.org/dc/elements/1.1/}title"):
            if t.text:
                lang = t.get(XML_LANG, "")
                if lang == "en" or not title:
                    title = t.text.strip()[:500]

        # Collect all vocabulary mappings: (vocab_id, field)
        mappings: list[tuple[str, str]] = []

        # Extract vocabulary references from CHO element
        # (XML tag, mapping field name)
        for xml_tag, field in CHO_VOCAB_FIELDS:
            for el in cho.findall(xml_tag):
                vid = extract_resource_ref(el)
                if vid:
                    mappings.append((vid, field))

        # Extract creator label and agent metadata from edm:Agent elements
        # Agents are siblings of ore:Aggregation inside rdf:RDF
        creator_label = ""
        for agent in metadata.iter("{http://www.europeana.eu/schemas/edm/}Agent"):
            agent_about = agent.get(RDF_ABOUT, "")
            if not any(f == "creator" and agent_about.endswith(v) for v, f in mappings):
                continue

            # Extract agent's name
            for pref_label in agent.findall("{http://www.w3.org/2004/02/skos/core#}prefLabel"):
                if pref_label.text:
                    lang = pref_label.get(XML_LANG, "")
                    if lang == "en" or not creator_label:
                        creator_label = pref_label.text.strip()

            # Extract birth/death place (rdaGr2 namespace, not edm)
            for bp in agent.findall("{http://rdvocab.info/ElementsGr2/}placeOfBirth"):
                vid = extract_resource_ref(bp)
                if vid:
                    mappings.append((vid, "birth_place"))
            for dp in agent.findall("{http://rdvocab.info/ElementsGr2/}placeOfDeath"):
                vid = extract_resource_ref(dp)
                if vid:
                    mappings.append((vid, "death_place"))

            # Extract profession/occupation
            for prof in agent.findall("{http://rdvocab.info/ElementsGr2/}professionOrOccupation"):
                vid = extract_resource_ref(prof)
                if vid:
                    mappings.append((vid, "profession"))

        # Add collection set mappings from header setSpec values
        for spec in set_specs:
            mappings.append((spec, "collection_set"))

        # Walk SKOS Concept / edm:Place / rdf:Description entity definitions
        # at the metadata root and capture their owl:sameAs URIs. This is the
        # exclusive channel for subject external IDs (Iconclass on dc:subject)
        # and the primary channel for spatial external IDs (TGN, GeoNames,
        # Wikidata on dcterms:spatial).
        ext_ids: list[tuple[str, str, str, str]] = []  # (vocab_id, authority, id, uri)
        for tag in (
            f"{{{NS['skos']}}}Concept",
            f"{{{NS['edm']}}}Place",
            f"{{{NS['rdf']}}}Description",
        ):
            for ent in metadata.iter(tag):
                ent_uri = ent.get(RDF_ABOUT, "")
                if not ent_uri:
                    continue
                vocab_id = ent_uri.rsplit("/", 1)[-1]
                for sa in ent.findall(OWL_SAME_AS):
                    sa_uri = sa.get(RDF_RESOURCE, "")
                    if not sa_uri:
                        continue
                    authority, local_id = classify_authority(sa_uri)
                    ext_ids.append((vocab_id, authority, local_id, sa_uri))

        # Multiple language variants joined with " | " (matches inscription /
        # provenance / credit_line convention from Phase 4).
        extent_parts: list[str] = []
        for ext in cho.findall("{http://purl.org/dc/terms/}extent"):
            if ext.text and ext.text.strip():
                extent_parts.append(ext.text.strip())
        extent_text = " | ".join(extent_parts) if extent_parts else None

        # Extract rights URI from ore:Aggregation
        rights_uri = ""
        agg = metadata.find(".//{http://www.openarchives.org/ore/terms/}Aggregation")
        if agg is not None:
            rights_el = agg.find("{http://www.europeana.eu/schemas/edm/}rights")
            if rights_el is not None:
                rights_uri = rights_el.get(RDF_RESOURCE, "")

        # Check for image availability and extract IIIF ID from edm:isShownBy / edm:object
        has_image = 0
        iiif_id = None
        if agg is not None:
            is_shown = agg.find("{http://www.europeana.eu/schemas/edm/}isShownBy")
            edm_obj = agg.find("{http://www.europeana.eu/schemas/edm/}object")

            # Extract IIIF URL from isShownBy (two RDF/XML shapes):
            #   1. <edm:isShownBy rdf:resource="https://iiif.micr.io/{UUID}/..."/>
            #   2. <edm:isShownBy><edm:WebResource rdf:about="https://iiif.micr.io/{UUID}/..."/></edm:isShownBy>
            iiif_url = ""
            if is_shown is not None:
                has_image = 1
                iiif_url = is_shown.get(RDF_RESOURCE, "")
                if not iiif_url:
                    # Nested WebResource — URL is on rdf:about of the child element
                    child = next(iter(is_shown), None)
                    if child is not None:
                        iiif_url = child.get(RDF_ABOUT, "")
            elif edm_obj is not None:
                has_image = 1
                iiif_url = edm_obj.get(RDF_RESOURCE, "")
                if not iiif_url:
                    child = next(iter(edm_obj), None)
                    if child is not None:
                        iiif_url = child.get(RDF_ABOUT, "")

            # Extract UUID from URL: https://iiif.micr.io/{UUID}/full/max/0/default.jpg
            if iiif_url:
                m = IIIF_ID_RE.match(iiif_url)
                if m:
                    iiif_id = m.group(1)

        records.append({
            "object_number": object_number,
            "title": title,
            "creator_label": creator_label,
            "rights_uri": rights_uri,
            "linked_art_uri": lod_uri,
            "has_image": has_image,
            "iiif_id": iiif_id,
            "mappings": mappings,
            "ext_ids": ext_ids,
            "extent_text": extent_text,
        })

    return records


def save_checkpoint(token: str, page: int):
    """Save harvest progress to checkpoint file."""
    with open(CHECKPOINT_PATH, "w") as f:
        json.dump({"resumption_token": token, "page": page}, f)


def load_checkpoint() -> tuple[str, int] | None:
    """Load harvest progress from checkpoint file."""
    if not CHECKPOINT_PATH.exists():
        return None
    try:
        with open(CHECKPOINT_PATH) as f:
            data = json.load(f)
        return data["resumption_token"], data["page"]
    except Exception:
        return None


def run_phase1(conn: sqlite3.Connection, resume: bool = False, max_pages: int | None = None):
    """Phase 1: Harvest OAI-PMH records. If max_pages is set, stop after that many pages (~200 records/page)."""
    start_page = 0
    url = f"{OAI_BASE}?verb=ListRecords&metadataPrefix=edm"

    if resume:
        checkpoint = load_checkpoint()
        if checkpoint:
            token, start_page = checkpoint
            url = f"{OAI_BASE}?verb=ListRecords&resumptionToken={token}"
            print(f"  Resuming from page {start_page + 1} (token: {token[:40]}...)")
        else:
            print("  No checkpoint found, starting fresh")

    page = start_page
    total_artworks = 0
    total_mappings = 0
    t0 = time.time()

    while url:
        page += 1
        try:
            root = fetch_oai_page(url)
        except Exception as e:
            print(f"  FATAL error on page {page}: {e}")
            print(f"  Use --resume to continue from last checkpoint")
            break

        records = extract_records(root)

        for rec in records:
            conn.execute(
                "INSERT OR IGNORE INTO artworks (object_number, title, creator_label, rights_uri, linked_art_uri, has_image, iiif_id, extent_text) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                (rec["object_number"], rec["title"], rec["creator_label"], rec["rights_uri"], rec["linked_art_uri"], rec["has_image"], rec["iiif_id"], rec.get("extent_text")),
            )
            for vocab_id, field in rec["mappings"]:
                conn.execute(
                    "INSERT OR IGNORE INTO mappings (object_number, vocab_id, field) VALUES (?, ?, ?)",
                    (rec["object_number"], vocab_id, field),
                )
                total_mappings += 1
            ext_ids = rec.get("ext_ids")
            if ext_ids:
                conn.executemany(VEI_INSERT_SQL, ext_ids)

        total_artworks += len(records)

        # Check for resumption token
        token_el = root.find(".//oai:resumptionToken", NS)
        if token_el is not None and token_el.text:
            token = token_el.text
            url = f"{OAI_BASE}?verb=ListRecords&resumptionToken={token}"
            # Save checkpoint
            save_checkpoint(token, page)
        else:
            url = None

        if max_pages and (page - start_page) >= max_pages:
            print(f"  Reached page limit ({max_pages} pages, {total_artworks:,} artworks)")
            url = None

        if page % 10 == 0:
            elapsed = time.time() - t0
            rate = (page - start_page) / elapsed * 60 if elapsed > 0 else 0
            print(
                f"  Page {page}: {total_artworks:,} artworks, {total_mappings:,} mappings "
                f"({rate:.0f} pages/min)",
                flush=True,
            )

        if page % BATCH_SIZE == 0:
            conn.commit()

    conn.commit()
    elapsed = time.time() - t0
    print(f"  Harvest complete: {total_artworks:,} artworks, {total_mappings:,} mappings, {page} pages, {elapsed:.0f}s")

    # Clean up checkpoint on success
    if CHECKPOINT_PATH.exists() and url is None:
        CHECKPOINT_PATH.unlink()


# ─── Phase 2: Resolve unmatched URIs ─────────────────────────────────

# Reasons the retry sweep will replay. MUST stay in sync with the return
# strings emitted by `resolve_uri` defined immediately below — specifically
# the three branches that raise on network/server faults rather than
# classify the response. Any new transient reason added to `resolve_uri`
# must be added here or it will be silently skipped by retries.
TRANSIENT_FAILURE_REASONS = frozenset({"timeout", "http_5xx", "unknown"})


def resolve_uri(entity_id: str) -> tuple[dict | None, str | None]:
    """Resolve a Rijksmuseum entity URI via the Linked Art API.

    Returns ``(result, reason)``. On success: ``(dict, None)``. On failure:
    ``(None, reason)`` where ``reason`` is one of: ``timeout``, ``http_404``,
    ``http_5xx``, ``http_<code>``, ``parse_error``, ``unsupported_type:<la_type>``,
    ``unknown``. Transient (``timeout``, ``http_5xx``, ``unknown``) vs permanent
    (``http_404``, ``unsupported_type:*``) is the axis `retry_transient_failures`
    uses to scope retries — see ``TRANSIENT_FAILURE_REASONS``.
    """
    url = f"{LINKED_ART_BASE}/{entity_id}"
    try:
        resp = get_http_session().get(url, headers={
            "Accept": "application/ld+json",
            "Profile": "https://linked.art/ns/v1/linked-art.json",
            "User-Agent": USER_AGENT,
        }, timeout=15)
    except (requests.Timeout, requests.ConnectionError):
        return None, "timeout"
    except requests.RequestException:
        return None, "unknown"

    if resp.status_code == 404:
        return None, "http_404"
    if 500 <= resp.status_code < 600:
        return None, "http_5xx"
    if not resp.ok:
        return None, f"http_{resp.status_code}"

    try:
        data = resp.json()
    except (json.JSONDecodeError, ValueError):
        return None, "parse_error"

    la_type = data.get("type", "")
    vocab_type = LA_TYPE_MAP.get(la_type)
    if not vocab_type:
        return None, f"unsupported_type:{la_type or 'missing'}"

    label_en = None
    label_nl = None
    name_variants = []  # All name variants for persons
    seen_names = set()  # Deduplicate (content, lang) pairs
    for name in data.get("identified_by", []):
        content = name.get("content", "")
        if not content or not isinstance(content, str):
            continue
        langs = name.get("language", [])
        lang_ids = [l.get("id", "") for l in langs] if langs else []
        if LANG_EN in lang_ids:
            label_en = label_en or content
        elif LANG_NL in lang_ids:
            label_nl = label_nl or content
        elif not label_en and not label_nl:
            label_en = content

        # Collect all name variants for persons (only Name entries, not Identifiers)
        if vocab_type == "person" and name.get("type") == "Name":
            lang = None
            if LANG_EN in lang_ids:
                lang = "en"
            elif LANG_NL in lang_ids:
                lang = "nl"

            classification = None
            for c in name.get("classified_as", []):
                cid = c.get("id", "")
                for suffix, clabel in AAT_NAME_CLASSIFICATION.items():
                    if cid.endswith(suffix):
                        classification = clabel
                        break
                if classification:
                    break

            key = (content, lang)
            if key not in seen_names:
                seen_names.add(key)
                name_variants.append({
                    "person_id": entity_id,
                    "name": content,
                    "lang": lang,
                    "classification": classification,
                })

    # #276: collect all equivalent[] authority IDs into _external_ids; the
    # single-value `external_id` column still gets the Wikidata-preferred
    # pick for back-compat.
    external_id = None
    ext_ids_seen: set[str] = set()
    external_ids_list: list[tuple[str, str, str]] = []
    for eq in data.get("equivalent", []):
        eq_id = eq.get("id", "")
        if not eq_id or eq_id in ext_ids_seen:
            continue
        ext_ids_seen.add(eq_id)
        authority, local_id = classify_authority(eq_id)
        external_ids_list.append((authority, local_id, eq_id))
        if "wikidata.org" in eq_id:
            external_id = eq_id  # Wikidata wins outright
        elif external_id is None:
            external_id = eq_id

    notation = None
    lat = None
    lon = None
    if vocab_type == "place":
        defined_by = data.get("defined_by", "")
        if isinstance(defined_by, str) and defined_by.startswith("POINT"):
            # Parse POINT(lon lat). #328: don't assign WKT to `notation` —
            # that column is reserved for Iconclass classifications.
            m = re.match(r"POINT\(([-\d.]+)\s+([-\d.]+)\)", defined_by)
            if m:
                lon = float(m.group(1))
                lat = float(m.group(2))

    return {
        "id": entity_id,
        "type": vocab_type,
        "label_en": label_en,
        "label_nl": label_nl,
        "external_id": external_id,
        "broader_id": None,
        "notation": notation,
        "lat": lat,
        "lon": lon,
        "name_variants": name_variants,
        "_external_ids": external_ids_list,
    }, None


def retry_transient_failures(conn: sqlite3.Connection, max_rounds: int = 3) -> int:
    """#225: Replay Phase 2 transient failures from the `phase2_failures` table.

    Called after the main Phase 2 / Phase 2b resolution loop completes. Loads
    all rows whose `reason` is in TRANSIENT_FAILURE_REASONS, retries them with
    exponential backoff (2s, 4s, 8s), upserts successes into the vocabulary
    table, removes recovered URIs from `phase2_failures`, and updates the
    reason column for URIs that transitioned from transient to permanent
    (e.g. a server that was 5xx is now 404).

    Permanent failures (404, parse_error, unsupported_type:*, http_4xx) are
    not touched — retrying them wastes requests and shifts no outcome.

    Returns the number of URIs successfully recovered.
    """
    rows = conn.execute(
        f"SELECT uri, reason FROM phase2_failures "
        f"WHERE reason IN ({','.join('?' * len(TRANSIENT_FAILURE_REASONS))})",
        tuple(TRANSIENT_FAILURE_REASONS),
    ).fetchall()
    if not rows:
        return 0

    # phase2_failures.uri always holds a bare Rijksmuseum entity ID — the main
    # run_phase2 loop inserts `eid` from `SELECT DISTINCT m.vocab_id`, which is
    # never URL-prefixed. No entity_id extraction needed.
    remaining: list[str] = [uri for uri, _reason in rows]
    recovered = 0
    print(f"  Retry sweep: {len(remaining)} transient failures to replay")

    # Retry rounds run in parallel within a round (HTTP in pool, SQL in main
    # thread via as_completed) and serialise across rounds so the backoff
    # sleep has any effect. Thread count is intentionally lower than
    # DEFAULT_THREADS — a retry sweep is second contact with a service we
    # just hammered, so be politer.
    RETRY_THREADS = min(6, DEFAULT_THREADS)
    PERSON_NAMES_INSERT = (
        "INSERT OR IGNORE INTO person_names "
        "(person_id, name, lang, classification) "
        "VALUES (:person_id, :name, :lang, :classification)"
    )

    for round_num in range(1, max_rounds + 1):
        if not remaining:
            break
        time.sleep(2 ** round_num)  # 2s, 4s, 8s
        print(f"    Round {round_num}/{max_rounds}: {len(remaining)} URIs", flush=True)

        next_remaining: list[str] = []
        recovered_this_round = 0
        with ThreadPoolExecutor(max_workers=RETRY_THREADS) as pool:
            futures = {pool.submit(resolve_uri, uri): uri for uri in remaining}
            for future in as_completed(futures):
                uri = futures[future]
                result, new_reason = future.result()
                if result:
                    ext_ids = result.pop("_external_ids", [])
                    variants = result.pop("name_variants", []) or []
                    conn.execute(VOCAB_INSERT_SQL, result)
                    if ext_ids:
                        conn.executemany(
                            VEI_INSERT_SQL,
                            [(result["id"], a, lid, u) for a, lid, u in ext_ids],
                        )
                    conn.execute("DELETE FROM phase2_failures WHERE uri = ?", (uri,))
                    if variants:
                        conn.executemany(PERSON_NAMES_INSERT, variants)
                    recovered += 1
                    recovered_this_round += 1
                elif new_reason in TRANSIENT_FAILURE_REASONS:
                    next_remaining.append(uri)
                else:
                    # Transitioned to permanent — pin the new reason so the
                    # next harvest doesn't retry it again.
                    conn.execute(
                        "UPDATE phase2_failures SET reason = ? WHERE uri = ?",
                        (new_reason or "unknown", uri),
                    )
        conn.commit()
        print(f"      Recovered {recovered_this_round}, {len(next_remaining)} still transient")
        remaining = next_remaining

    return recovered


def sync_phase_failures(conn: sqlite3.Connection) -> None:
    """#226: Mirror persistent Phase 2 failures into the unified `phase_failures`
    table so post-harvest diagnostics have a stable query surface after Phase 3
    integer-encoding drops `tier2_done` and other resume markers.

    Called at end of `run_phase2` after `retry_transient_failures` has already
    removed recovered URIs — everything left in `phase2_failures` is either
    permanent (404, unsupported_type, parse_error) or persistently-transient
    (still failing after 3 retry rounds). Both are worth preserving.
    """
    conn.execute(
        """
        INSERT OR REPLACE INTO phase_failures
            (phase, uri_or_objnum, failure_type, error_message, retry_count)
        SELECT 2, uri, reason, NULL, 0 FROM phase2_failures
        """
    )
    conn.commit()


def run_phase2(conn: sqlite3.Connection):
    """Phase 2: Resolve all unmapped vocabulary URIs."""
    cur = conn.cursor()

    # Integer-encoded DBs have vocab_rowid (always a valid FK) — nothing to resolve
    if "vocab_rowid" in get_columns(conn, "mappings"):
        print("  Integer-encoded schema: all vocab references are valid FKs, nothing to resolve.")
        unmatched = []
    else:
        unmatched = [row[0] for row in cur.execute("""
            SELECT DISTINCT m.vocab_id
            FROM mappings m
            LEFT JOIN vocabulary v ON m.vocab_id = v.id
            WHERE v.id IS NULL
        """).fetchall()]

    if not unmatched:
        print("  No unmatched vocabulary URIs to resolve.")
        return

    print(f"  Resolving {len(unmatched):,} unmatched vocabulary URIs ({DEFAULT_THREADS} threads)...")

    PERSON_NAMES_INSERT = (
        "INSERT OR IGNORE INTO person_names (person_id, name, lang, classification) "
        "VALUES (:person_id, :name, :lang, :classification)"
    )

    resolved = 0
    person_names_count = 0
    t0 = time.time()

    FAILURES_INSERT = (
        "INSERT OR REPLACE INTO phase2_failures (uri, reason) VALUES (?, ?)"
    )
    failure_batch: list[tuple[str, str]] = []
    failures_by_reason: Counter[str] = Counter()

    with ThreadPoolExecutor(max_workers=DEFAULT_THREADS) as pool:
        futures = {pool.submit(resolve_uri, eid): eid for eid in unmatched}
        batch = []
        name_batch = []

        for i, future in enumerate(as_completed(futures), 1):
            eid = futures[future]
            result, reason = future.result()
            if result:
                batch.append(result)
                variants = result.get("name_variants", [])
                if variants:
                    name_batch.extend(variants)
                    person_names_count += len(variants)
                resolved += 1
            else:
                key = reason or "unknown"
                failure_batch.append((eid, key))
                failures_by_reason[key] += 1

            if len(batch) >= 200:
                vei_rows: list[tuple[str, str, str, str]] = []
                for r in batch:
                    for a, lid, u in r.pop("_external_ids", []):
                        vei_rows.append((r["id"], a, lid, u))
                conn.executemany(VOCAB_INSERT_SQL, batch)
                if vei_rows:
                    conn.executemany(VEI_INSERT_SQL, vei_rows)
                if name_batch:
                    conn.executemany(PERSON_NAMES_INSERT, name_batch)
                    name_batch = []
                if failure_batch:
                    conn.executemany(FAILURES_INSERT, failure_batch)
                    failure_batch = []
                conn.commit()
                batch = []

            if i % 1000 == 0:
                elapsed = time.time() - t0
                rate = i / elapsed
                remaining = (len(unmatched) - i) / rate
                failed = sum(failures_by_reason.values())
                print(
                    f"  {i:,}/{len(unmatched):,} ({resolved:,} ok, {failed:,} failed, "
                    f"{rate:.0f}/s, ~{remaining:.0f}s left)",
                    flush=True,
                )

    if batch:
        vei_rows = []
        for r in batch:
            for a, lid, u in r.pop("_external_ids", []):
                vei_rows.append((r["id"], a, lid, u))
        conn.executemany(VOCAB_INSERT_SQL, batch)
        if vei_rows:
            conn.executemany(VEI_INSERT_SQL, vei_rows)
    if name_batch:
        conn.executemany(PERSON_NAMES_INSERT, name_batch)
    if failure_batch:
        conn.executemany(FAILURES_INSERT, failure_batch)
    conn.commit()

    failed = sum(failures_by_reason.values())
    elapsed = time.time() - t0
    print(f"  Resolution complete: {resolved:,} resolved, {failed:,} failed, {elapsed:.0f}s")
    if person_names_count:
        print(f"  Person name variants collected: {person_names_count:,}")
    if failures_by_reason:
        breakdown = ", ".join(
            f"{reason}={count:,}" for reason, count in failures_by_reason.most_common()
        )
        print(f"  Failure breakdown: {breakdown}")
        print(f"  → {failed:,} URIs recorded in phase2_failures table for backfill review")

    # #225: retry sweep for transient failures. Runs before sync_phase_failures
    # so recovered URIs are removed from phase2_failures before the mirror copy.
    recovered = retry_transient_failures(conn)
    if recovered:
        print(f"  Retry sweep recovered {recovered:,} previously-failed URIs")

    # #226: mirror persistent failures into the unified phase_failures table
    # so they survive Phase 3 integer-encoding for post-harvest diagnostics.
    sync_phase_failures(conn)


# ─── Phase 4: Linked Art Resolution (Tier 2) ─────────────────────────

def has_classification(classified_as: list | None, uris: str | set[str]) -> bool:
    """Check if a classified_as array contains any of the given URIs."""
    if not classified_as:
        return False
    if isinstance(uris, str):
        uris = {uris}
    return any(
        (c.get("id", "") if isinstance(c, dict) else str(c)) in uris
        for c in classified_as
    )


def find_statements(referred_to_by: list | None, aat_uri: str) -> list[str]:
    """Find all referred_to_by statements matching a given AAT classification.

    Returns all matching texts concatenated with ' | ' delimiter.
    Collects ALL languages — inscriptions are often in Latin/Dutch/mixed.
    """
    if not referred_to_by:
        return []
    texts = []
    for stmt in referred_to_by:
        if not isinstance(stmt, dict):
            continue
        if has_classification(stmt.get("classified_as"), aat_uri):
            content = stmt.get("content", "")
            if isinstance(content, list):
                texts.extend(s for s in content if isinstance(s, str))
            elif content:
                texts.append(content)
    return texts


# Conversion factors from unit to centimeters
UNIT_TO_CM = {
    AAT_UNIT_CM: 1.0,
    AAT_UNIT_MM: 0.1,
    AAT_UNIT_M: 100.0,
}

# Conversion factors from unit to grams (for weight)
UNIT_TO_G = {
    AAT_UNIT_G: 1.0,
    AAT_UNIT_KG: 1000.0,
}


def _extract_dimension_value(dimensions: list | None, type_uris: set[str], unit_map: dict[str, float]) -> float | None:
    """Extract a dimension/weight value, converting via unit_map (default factor 1.0)."""
    if not dimensions:
        return None
    for dim in dimensions:
        if not isinstance(dim, dict):
            continue
        value = dim.get("value")
        if value is None:
            continue
        if not has_classification(dim.get("classified_as", []), type_uris):
            continue
        unit = dim.get("unit", {})
        unit_id = unit.get("id", "") if isinstance(unit, dict) else ""
        try:
            val = float(value)
        except (ValueError, TypeError):
            continue
        factor = unit_map.get(unit_id, 1.0)
        return round(val * factor, 2) if factor != 1.0 else val
    return None


def extract_dimension_cm(dimensions: list | None, type_uris: set[str]) -> float | None:
    """Extract a dimension value in centimeters for a given dimension type (height/width)."""
    return _extract_dimension_value(dimensions, type_uris, UNIT_TO_CM)


def extract_weight_g(dimensions: list | None, type_uris: set[str]) -> float | None:
    """Extract a weight value in grams for a given dimension type."""
    return _extract_dimension_value(dimensions, type_uris, UNIT_TO_G)


def extract_dimension_note(dimensions: list | None) -> str | None:
    """Extract annotation text from dimension referred_to_by entries."""
    if not dimensions:
        return None
    notes = []
    for dim in dimensions:
        if not isinstance(dim, dict):
            continue
        for ref in dim.get("referred_to_by", []):
            if not isinstance(ref, dict):
                continue
            content = ref.get("content", "")
            if isinstance(content, list):
                notes.extend(s for s in content if isinstance(s, str))
            elif content:
                notes.append(content)
    return " | ".join(notes) if notes else None


def _extract_timespan(entry: dict) -> tuple[str | None, str | None, str | None]:
    """Extract (date_display, date_begin, date_end) from a Linked Art timespan dict."""
    ts = entry.get("timespan", {})
    if not isinstance(ts, dict):
        return None, None, None
    date_begin = ts.get("begin_of_the_begin")
    date_end = ts.get("end_of_the_end")
    date_display = None
    for ident in ts.get("identified_by", []):
        if isinstance(ident, dict) and ident.get("type") == "Name":
            content = ident.get("content", "")
            if content:
                date_display = content
                break
    return date_display, date_begin, date_end


def extract_modifications(data: dict) -> list[dict]:
    """Extract modification/treatment events from modified_by.

    Returns list of dicts: {modifier_uri, date_display, date_begin, date_end, description}
    """
    modified_by = data.get("modified_by", [])
    if not isinstance(modified_by, list):
        return []
    results = []
    for entry in modified_by:
        if not isinstance(entry, dict):
            continue
        # carried_out_by: bare URI strings
        modifier_uri = None
        for actor in entry.get("carried_out_by", []):
            if isinstance(actor, str):
                modifier_uri = actor
                break
            elif isinstance(actor, dict) and actor.get("id"):
                modifier_uri = actor["id"]
                break

        date_display, date_begin, date_end = _extract_timespan(entry)

        # description from referred_to_by (prefer EN via AAT 300388277)
        description = None
        for ref in entry.get("referred_to_by", []):
            if not isinstance(ref, dict):
                continue
            content = ref.get("content", "")
            if isinstance(content, list):
                content = " ".join(s for s in content if isinstance(s, str))
            if content:
                description = content
                break

        results.append({
            "modifier_uri": modifier_uri,
            "date_display": date_display,
            "date_begin": date_begin,
            "date_end": date_end,
            "description": description,
        })
    return results


def extract_attributed_by(data: dict) -> tuple[list[dict], list[dict]]:
    """Extract attributed_by entries, splitting into related_objects and examinations.

    Pattern A (no carried_out_by): Object relationships → related_objects
    Pattern B (has carried_out_by): Examination reports → examinations

    Returns (related_objects_list, examinations_list).
    """
    attributed_by = data.get("attributed_by", [])
    if not isinstance(attributed_by, list):
        return [], []

    related_objects = []
    examinations = []

    for entry in attributed_by:
        if not isinstance(entry, dict):
            continue

        has_examiner = bool(entry.get("carried_out_by"))

        if has_examiner:
            # Pattern B: Examination report
            examiner_name = None
            for actor in entry.get("carried_out_by", []):
                if isinstance(actor, dict):
                    # Try to get label from identified_by
                    for ident in actor.get("identified_by", []):
                        if isinstance(ident, dict):
                            name = ident.get("content", "")
                            if name:
                                examiner_name = name
                                break
                    if not examiner_name and actor.get("id"):
                        examiner_name = actor["id"].split("/")[-1]

            report_type_id = ""
            report_type_en = None
            for cls in entry.get("classified_as", []):
                if isinstance(cls, dict) and cls.get("id"):
                    report_type_id = cls["id"]
                    # The label lives in notation[] (JSON-LD expanded form:
                    # {"@language": "en"/"nl", "@value": "..."}), English
                    # preferred, Dutch fallback, then any. (The older
                    # identified_by[].content shape — used by the dereferenced
                    # concept record — is kept as a last resort.)
                    report_type_nl = None
                    report_type_any = None
                    for note in cls.get("notation", []):
                        if not isinstance(note, dict):
                            continue
                        value = note.get("@value")
                        if not value:
                            continue
                        lang = note.get("@language")
                        if lang == "en":
                            report_type_en = value
                        elif lang == "nl":
                            report_type_nl = value
                        elif report_type_any is None:
                            report_type_any = value
                    if not report_type_en:
                        for ident in cls.get("identified_by", []):
                            if isinstance(ident, dict) and ident.get("content"):
                                report_type_en = ident.get("content")
                                break
                    report_type_en = report_type_en or report_type_nl or report_type_any
                    break

            date_display, date_begin, date_end = _extract_timespan(entry)

            if report_type_id:
                examinations.append({
                    "examiner_name": examiner_name,
                    "report_type_id": report_type_id,
                    "report_type_en": report_type_en,
                    "date_display": date_display,
                    "date_begin": date_begin,
                    "date_end": date_end,
                })
        else:
            # Pattern A: Object relationships
            for assigned in entry.get("assigned", []):
                if not isinstance(assigned, dict):
                    continue
                related_uri = assigned.get("id", "")
                if not related_uri:
                    continue

                # Extract relationship labels from identified_by
                relationship_en = None
                relationship_nl = None
                for ident in entry.get("identified_by", []):
                    if not isinstance(ident, dict):
                        continue
                    content = ident.get("content", "")
                    if not content:
                        continue
                    lang_uri = None
                    for lang in ident.get("language", []):
                        if isinstance(lang, dict):
                            lang_uri = lang.get("id", "")
                            break
                    if lang_uri == LANG_EN:
                        relationship_en = content
                    elif lang_uri == LANG_NL:
                        relationship_nl = content
                    elif not relationship_en:
                        relationship_en = content  # fallback

                if relationship_en:
                    related_objects.append({
                        "related_la_uri": related_uri,
                        "relationship_en": relationship_en,
                        "relationship_nl": relationship_nl,
                    })

    return related_objects, examinations


def extract_title_variants(data: dict) -> list[dict]:
    """Extract title variants from identified_by Name entries.

    Returns list of dicts: {title_text, language, qualifier}
    """
    results = []
    for entry in data.get("identified_by", []):
        if not isinstance(entry, dict) or entry.get("type") != "Name":
            continue
        content = entry.get("content", "")
        if isinstance(content, list):
            content = " ".join(s for s in content if isinstance(s, str))
        if not content:
            continue

        # Language
        language = None
        for lang in entry.get("language", []):
            if isinstance(lang, dict):
                lid = lang.get("id", "")
                if lid == LANG_EN:
                    language = "en"
                elif lid == LANG_NL:
                    language = "nl"
                elif lid:
                    language = lid.split("/")[-1]
                break

        # Qualifier from classified_as
        qualifier = None
        for cls in entry.get("classified_as", []):
            if isinstance(cls, dict):
                cid = cls.get("id", "")
                if cid == AAT_TITLE_FULL:
                    qualifier = "full"
                elif cid == AAT_TITLE_BRIEF:
                    qualifier = "brief"
                elif cid == AAT_TITLE_DISPLAY:
                    qualifier = "display"
                elif cid == RM_TITLE_FORMER:
                    qualifier = "former"
                if qualifier:
                    break

        results.append({
            "title_text": content,
            "language": language,
            "qualifier": qualifier,
        })
    return results


def extract_part_of(data: dict) -> list[str]:
    """Extract parent HMO URIs from part_of.

    Returns list of Linked Art URIs.
    """
    results = []
    for entry in data.get("part_of", []):
        if isinstance(entry, dict) and entry.get("type") == "HumanMadeObject":
            uri = entry.get("id", "")
            if uri:
                results.append(uri)
    return results


# ─── v0.26 extractors ────────────────────────────────────────────────

def extract_record_timestamps(data: dict) -> tuple[str | None, str | None]:
    """SHIP-1: aggregate record-edit timestamps across referred_to_by[].

    Each text statement carries a `created_by.timespan` describing when
    it was authored. We take the earliest begin_of_the_begin (record's
    first known creation event) and the latest end_of_the_end (last
    known modification event). Both are ISO 8601 strings or None.
    """
    begins: list[str] = []
    ends: list[str] = []
    for ref in data.get("referred_to_by", []):
        if not isinstance(ref, dict):
            continue
        cb = ref.get("created_by")
        if not isinstance(cb, dict):
            continue
        ts = cb.get("timespan")
        if not isinstance(ts, dict):
            continue
        b = ts.get("begin_of_the_begin")
        e = ts.get("end_of_the_end")
        if isinstance(b, str) and b:
            begins.append(b)
        if isinstance(e, str) and e:
            ends.append(e)
    record_created = min(begins) if begins else None
    record_modified = max(ends) if ends else None
    return record_created, record_modified


def _ref_id(ref) -> str | None:
    """Return the URI from a Linked Art reference — accepts a bare URI string
    or a dict with `id`. Production payloads use both shapes interchangeably
    inside `classified_as` / `carried_by` arrays."""
    if isinstance(ref, str):
        return ref or None
    if isinstance(ref, dict):
        rid = ref.get("id", "")
        return rid or None
    return None


def extract_attribution_evidence(data: dict) -> list[dict]:
    """SHIP-2: walk produced_by.part[].assigned_by[].motivated_by[] to capture
    the S5 attribution-evidence trail.

    Returns list of dicts: {part_index, evidence_type_aat, carried_by_uri, label_text}.
    Multiple rows per artwork are normal; the schema's PRIMARY KEY
    (art_id, part_index, evidence_type_aat, carried_by_uri) deduplicates.
    """
    results: list[dict] = []
    produced_by = data.get("produced_by")
    if not isinstance(produced_by, dict):
        return results
    parts = produced_by.get("part", [])
    if not isinstance(parts, list):
        return results
    for part_index, part in enumerate(parts):
        if not isinstance(part, dict):
            continue
        for assigned in part.get("assigned_by", []):
            if not isinstance(assigned, dict):
                continue
            for motivated in assigned.get("motivated_by", []):
                if not isinstance(motivated, dict):
                    continue
                # Evidence type AAT (first id from classified_as[]).
                # Items may be bare strings or dicts.
                evidence_type = None
                for cls in motivated.get("classified_as", []):
                    cid = _ref_id(cls)
                    if cid:
                        evidence_type = cid
                        break
                # Each carried_by[] item is one publication/citation URI;
                # may be bare strings or dicts.
                carried_uris = [_ref_id(cb) for cb in motivated.get("carried_by", [])]
                carried_uris = [u for u in carried_uris if u]
                if not carried_uris:
                    carried_uris = [None]  # at least one row even without a URI
                # Free-text label (prefer EN, fallback any).
                label_by_lang: dict[str, str] = {}
                for ident in motivated.get("identified_by", []):
                    if not isinstance(ident, dict):
                        continue
                    content = ident.get("content", "")
                    if isinstance(content, list):
                        content = " ".join(s for s in content if isinstance(s, str))
                    if not content:
                        continue
                    lang_uri = ""
                    for lang in ident.get("language", []):
                        if isinstance(lang, dict):
                            lang_uri = lang.get("id", "")
                            break
                    if lang_uri and lang_uri not in label_by_lang:
                        label_by_lang[lang_uri] = content
                    elif "" not in label_by_lang:
                        label_by_lang[""] = content
                label_text = _pick_preferred_lang(label_by_lang)
                for carried_uri in carried_uris:
                    results.append({
                        "part_index": part_index,
                        "evidence_type_aat": evidence_type,
                        "carried_by_uri": carried_uri,
                        "label_text": label_text,
                    })
    return results


def extract_about(data: dict) -> list[str]:
    """SHIP-5: thematic vocab links from la-framed `about[]`.

    Returns list of vocab_id strings (the URI suffix). Caller writes
    rows into mappings(field='theme'). Unresolved entries are silently
    dropped at INSERT time per existing orphan-mapping policy.

    Note: HumanMadeObject payloads do NOT carry `about[]` directly in
    production data — `about[]` lives on the VisualItem reached via
    `shows[0].id`. See `extract_visualitem_uri` and Phase 4.5.
    """
    results: list[str] = []
    for entry in data.get("about", []):
        uri = _ref_id(entry)
        if not uri:
            continue
        results.append(uri.rsplit("/", 1)[-1])
    return results


def extract_visualitem_uri(data: dict) -> str | None:
    """Return the VisualItem URI from an HMO's `shows[]` array.

    Production HMOs carry exactly one `shows[]` entry pointing at the
    VisualItem (the work's image-bearing aspect). Returns None when no
    shows[] entry exists or the first entry has no id.
    """
    for entry in data.get("shows", []):
        uri = _ref_id(entry)
        if uri:
            return uri
    return None


def extract_artwork_external_ids(data: dict) -> list[tuple[str, str, str]]:
    """SHIP-7: artwork-level sameAs/equivalent URIs (Wikidata, handle, etc.).

    Reads `equivalent[].id` on the HMO root. For VisualItem-rooted
    payloads (some shape variants), additionally reads
    `shown_by[].equivalent[].id`.

    Returns list of (authority, local_id, uri) tuples ready to insert
    into artwork_external_ids. classify_authority() handles bucketing.
    """
    results: list[tuple[str, str, str]] = []
    seen: set[tuple[str, str]] = set()

    def _emit(uri: str) -> None:
        if not uri:
            return
        authority, local_id = classify_authority(uri)
        key = (authority, local_id)
        if key in seen:
            return
        seen.add(key)
        results.append((authority, local_id, uri))

    for eq in data.get("equivalent", []):
        _emit(_ref_id(eq) or "")

    for shown in data.get("shown_by", []):
        if not isinstance(shown, dict):
            continue
        for eq in shown.get("equivalent", []):
            _emit(_ref_id(eq) or "")

    return results


def _pick_preferred_lang(by_lang: dict[str, str]) -> str | None:
    """Return EN value, then NL, then first available."""
    return by_lang.get(LANG_EN) or by_lang.get(LANG_NL) or (
        next(iter(by_lang.values())) if by_lang else None
    )


def _extract_ids(items: list, field: str) -> list[tuple[str, str]]:
    """Extract (vocab_id, field) tuples from a list of Linked Art typed dicts."""
    result = []
    for item in items:
        if isinstance(item, dict):
            uri = item.get("id", "")
            if uri:
                result.append((uri.split("/")[-1], field))
    return result


def _extract_assigned_creators(items: list) -> list[tuple[str, str]]:
    """Extract creator (vocab_id, field) tuples from assigned[] items.

    Handles two patterns:
      1. Direct person: {"type": "Person", "id": "https://.../{id}"}
      2. Inline Group (workshop of): {"type": "Group", "formed_by": {
             "influenced_by": [{"type": "Person", "id": "https://.../{id}"}]}}
    """
    result = []
    for item in items:
        if not isinstance(item, dict):
            continue
        uri = item.get("id", "")
        if uri:
            result.append((uri.split("/")[-1], "creator"))
            continue
        # Inline Group — traverse formed_by.influenced_by to find the person
        if item.get("type") == "Group":
            formed_by = item.get("formed_by")
            if isinstance(formed_by, dict):
                for inf in formed_by.get("influenced_by", []):
                    if isinstance(inf, dict) and inf.get("id"):
                        result.append((inf["id"].split("/")[-1], "creator"))
    return result


def extract_production_parts(data: dict) -> tuple[
    list[tuple[str, str]], list[tuple[str, str]], list[tuple[str, str]],
    list[tuple[str, str, int]], list[tuple[str, str]], list[tuple[str, str]],
]:
    """Extract production roles, qualifiers, creators, assignment pairs, places, and source types from produced_by.

    Returns:
        (roles, qualifiers, creators, assignment_pairs, places, source_types) where
        roles/qualifiers/creators/places/source_types are lists of (vocab_id, field) tuples,
        and assignment_pairs are (qualifier_id, creator_id, part_index) tuples.

    Production structure in Linked Art:
        produced_by: {
            part: [
                {
                    carried_out_by: [{ id: "https://id.rijksmuseum.nl/31xxx" }],
                    technique: [{ id: "https://id.rijksmuseum.nl/12xxx" }],  # role (painter, printmaker)
                    classified_as: [{ id: "..." }],  # priority level (primary/secondary/undetermined)
                    took_place_at: [{ id: "https://id.rijksmuseum.nl/230xxx", type: "Place" }],
                    used_object_of_type: [{ id: "http://vocab.getty.edu/aat/300xxx", type: "Type" }],
                    assigned_by: [
                        {
                            type: "AttributeAssignment",
                            assigned_property: "carried_out_by",
                            assigned: [{ id: "https://id.rijksmuseum.nl/31xxx" }],  # person
                            classified_as: [{ id: "..." }],  # REAL qualifier (attributed to, workshop of)
                        }
                    ]
                }
            ]
        }

    Note: part.classified_as contains priority levels (primary/secondary), NOT rich
    attribution qualifiers. The real qualifiers (attributed to, workshop of, manner of,
    etc.) come from assigned_by[].classified_as. Both are stored as
    "attribution_qualifier" mappings (different AAT codes, additive).
    """
    roles: list[tuple[str, str]] = []
    qualifiers: list[tuple[str, str]] = []
    creators: list[tuple[str, str]] = []
    assignment_pairs: list[tuple[str, str, int]] = []  # (qualifier_id, creator_id, part_index)
    places: list[tuple[str, str]] = []
    source_types: list[tuple[str, str]] = []

    produced_by = data.get("produced_by")
    if not isinstance(produced_by, dict):
        return roles, qualifiers, creators, assignment_pairs, places, source_types

    # Collect parts from both part[] and top-level assigned_by
    parts = produced_by.get("part", [])
    if not isinstance(parts, list):
        # Single production event without parts — check top level
        parts = [produced_by]

    def _process_assigned_by(assigned_by_list: list, part_idx: int):
        """Process assigned_by entries, extracting creators, qualifiers, and pairs."""
        for assignment in assigned_by_list:
            if not isinstance(assignment, dict):
                continue
            if assignment.get("type") != "AttributeAssignment":
                continue
            prop = assignment.get("assigned_property", "")
            if prop not in ("carried_out_by", "influenced_by"):
                continue
            assigned_creators = _extract_assigned_creators(assignment.get("assigned", []))
            assigned_quals = _extract_ids(assignment.get("classified_as", []), "attribution_qualifier")
            creators.extend(assigned_creators)
            qualifiers.extend(assigned_quals)

            # Emit assignment_pairs: pair each qualifier with each creator
            for q_id, _ in assigned_quals:
                for c_id, _ in assigned_creators:
                    assignment_pairs.append((q_id, c_id, part_idx))

    def _normalise_list(val: object) -> list:
        """Normalise a value that may be a dict, list, or absent into a list."""
        if isinstance(val, dict):
            return [val]
        if isinstance(val, list):
            return val
        return []

    for part_idx, part in enumerate(parts):
        if not isinstance(part, dict):
            continue
        roles.extend(_extract_ids(part.get("technique", []), "production_role"))
        qualifiers.extend(_extract_ids(part.get("classified_as", []), "attribution_qualifier"))
        _process_assigned_by(part.get("assigned_by", []), part_idx)
        places.extend(_extract_ids(_normalise_list(part.get("took_place_at")), "production_place"))
        source_types.extend(_extract_ids(_normalise_list(part.get("used_object_of_type")), "source_type"))

    # Also check top-level produced_by.assigned_by (#43 — some artworks have it at the top)
    if produced_by.get("part") is not None:
        # Only process top-level assigned_by if we didn't already treat produced_by as a single part
        _process_assigned_by(produced_by.get("assigned_by", []), -1)
        # Top-level took_place_at / used_object_of_type (rare but possible)
        places.extend(_extract_ids(_normalise_list(produced_by.get("took_place_at")), "production_place"))
        source_types.extend(_extract_ids(_normalise_list(produced_by.get("used_object_of_type")), "source_type"))

    return roles, qualifiers, creators, assignment_pairs, places, source_types


def extract_narrative(data: dict) -> str | None:
    """Extract curatorial narrative text from Linked Art subject_of.

    Walks subject_of[] entries looking for part[] children classified as
    AAT 300048722 (essay). Prefers EN, falls back to NL, then any language.
    """
    subject_of = data.get("subject_of")
    if not isinstance(subject_of, list):
        return None

    narratives: dict[str, str] = {}  # keyed by lang URI

    for entry in subject_of:
        if not isinstance(entry, dict):
            continue
        # Determine language of this entry
        lang_uri = None
        for lang in entry.get("language", []):
            lid = lang.get("id", "") if isinstance(lang, dict) else ""
            if lid in (LANG_NL, LANG_EN):
                lang_uri = lid
                break
            if lid and not lang_uri:
                lang_uri = lid

        # Walk parts looking for narrative classification
        for part in entry.get("part", []):
            if not isinstance(part, dict):
                continue
            if has_classification(part.get("classified_as"), AAT_NARRATIVE):
                content = part.get("content", "")
                if content and lang_uri and lang_uri not in narratives:
                    narratives[lang_uri] = content

    return _pick_preferred_lang(narratives)


def resolve_artwork(uri: str) -> dict | None:
    """Resolve a single artwork's Linked Art JSON-LD and extract Tier 2 fields."""
    try:
        resp = get_http_session().get(uri, headers={
            "Accept": "application/ld+json",
            "Profile": "https://linked.art/ns/v1/linked-art.json",
            "User-Agent": USER_AGENT,
        }, timeout=30)
    except requests.RequestException as e:
        print(f"    Error for {uri}: {e}", flush=True)
        return None

    if resp.status_code == 404:
        return {"_status": "not_found"}
    if not resp.ok:
        # Log non-404 HTTP errors for diagnostics (500s, 503s, rate limits)
        print(f"    HTTP {resp.status_code} for {uri}", flush=True)
        return None

    try:
        data = resp.json()
    except ValueError as e:
        print(f"    Error for {uri}: {e}", flush=True)
        return None

    referred_to_by = data.get("referred_to_by", [])

    # Extract text fields by AAT classification
    def join_statements(aat_uri: str) -> str | None:
        texts = find_statements(referred_to_by, aat_uri)
        return " | ".join(texts) if texts else None

    inscription_text = join_statements(AAT_INSCRIPTIONS)
    provenance_text = join_statements(AAT_PROVENANCE)
    credit_line = join_statements(AAT_CREDIT_LINE)
    description_text = join_statements(AAT_DESCRIPTION)

    # Structured dimensions
    dimensions = data.get("dimension", [])
    height_cm = extract_dimension_cm(dimensions, HEIGHT_URIS)
    width_cm = extract_dimension_cm(dimensions, WIDTH_URIS)
    depth_cm = extract_dimension_cm(dimensions, DEPTH_URIS)
    diameter_cm = extract_dimension_cm(dimensions, DIAMETER_URIS)
    weight_g = extract_weight_g(dimensions, WEIGHT_URIS)
    dimension_note = extract_dimension_note(dimensions)

    # Production roles, attribution qualifiers, assigned creators, assignment pairs, places, source types
    roles, qualifiers, creators, assignment_pairs, places, source_types = extract_production_parts(data)

    # Creator label from production statement text (referred_to_by with AAT 300435416)
    # Contains the qualifier-prefixed creator name (e.g. "attributed to Claes van Beresteyn")
    creator_label = None
    produced_by = data.get("produced_by", {})
    if isinstance(produced_by, dict):
        prod_refs = produced_by.get("referred_to_by", [])
        if isinstance(prod_refs, list):
            label_by_lang: dict[str, str] = {}
            for ref in prod_refs:
                if not isinstance(ref, dict):
                    continue
                if not has_classification(ref.get("classified_as"), AAT_PRODUCTION_STATEMENT):
                    continue
                content = ref.get("content", "")
                if isinstance(content, list):
                    content = " ".join(s for s in content if isinstance(s, str))
                if not content:
                    continue
                for lang in ref.get("language", []):
                    lid = lang.get("id", "") if isinstance(lang, dict) else ""
                    if lid and lid not in label_by_lang:
                        label_by_lang[lid] = content
                        break
            creator_label = _pick_preferred_lang(label_by_lang)

    # Curatorial narrative (museum wall text)
    narrative_text = extract_narrative(data)

    # Date extraction from produced_by.timespan
    date_earliest = None
    date_latest = None
    produced_by = data.get("produced_by", {})
    timespan = produced_by.get("timespan", {}) if isinstance(produced_by, dict) else {}
    if isinstance(timespan, list) and timespan:
        # Multi-phase production: take widest date range across all phases
        all_begins = [t.get("begin_of_the_begin", "") for t in timespan if isinstance(t, dict)]
        all_ends = [t.get("end_of_the_end", "") for t in timespan if isinstance(t, dict)]
        timespan = {
            "begin_of_the_begin": min((b for b in all_begins if b), default=""),
            "end_of_the_end": max((e for e in all_ends if e), default=""),
        }
    if isinstance(timespan, dict):
        for key, target in [("begin_of_the_begin", "earliest"), ("end_of_the_end", "latest")]:
            val = timespan.get(key, "")
            if val and isinstance(val, str) and len(val) >= 4:
                try:
                    # Handle negative years (BCE) and standard ISO dates
                    year_str = val[:5] if val.startswith("-") else val[:4]
                    year = int(year_str)
                    if target == "earliest":
                        date_earliest = year
                    else:
                        date_latest = year
                except (ValueError, IndexError):
                    pass
    # If only one is present, use it for both (single-year works)
    if date_earliest is not None and date_latest is None:
        date_latest = date_earliest
    elif date_latest is not None and date_earliest is None:
        date_earliest = date_latest

    # Title extraction: all Name entries from identified_by
    title_parts = []
    for entry in data.get("identified_by", []):
        if isinstance(entry, dict) and entry.get("type") == "Name":
            content = entry.get("content", "")
            if isinstance(content, list):
                title_parts.extend(s for s in content if isinstance(s, str))
            elif content:
                title_parts.append(content)
    title_all_text = "\n".join(title_parts) if title_parts else None

    # Date display text from timespan.identified_by (reuse produced_by from above)
    date_display = None
    ts_for_display = produced_by.get("timespan", {}) if isinstance(produced_by, dict) else {}
    if isinstance(ts_for_display, list):
        # Multi-phase: concatenate display labels with "; "
        display_parts = []
        for ts_item in ts_for_display:
            if isinstance(ts_item, dict):
                for ident in ts_item.get("identified_by", []):
                    if isinstance(ident, dict) and ident.get("type") == "Name":
                        c = ident.get("content", "")
                        if c:
                            display_parts.append(c)
                            break
        date_display = "; ".join(display_parts) if display_parts else None
    elif isinstance(ts_for_display, dict):
        for ident in ts_for_display.get("identified_by", []):
            if isinstance(ident, dict) and ident.get("type") == "Name":
                c = ident.get("content", "")
                if c:
                    date_display = c
                    break

    # Current location
    current_location = None
    cur_loc = data.get("current_location", {})
    if isinstance(cur_loc, dict):
        # Try label from identified_by, then fall back to id
        for ident in cur_loc.get("identified_by", []):
            if isinstance(ident, dict):
                c = ident.get("content", "")
                if c:
                    current_location = c
                    break
        if not current_location:
            current_location = cur_loc.get("id")

    # Provenance text hash (SHA-256 for incremental change detection)
    provenance_text_hash = None
    if provenance_text:
        provenance_text_hash = hashlib.sha256(provenance_text.encode("utf-8")).hexdigest()

    # Modifications (treatments/restorations)
    modifications = extract_modifications(data)

    # Related objects and examinations from attributed_by
    related_objects, examinations = extract_attributed_by(data)

    # Title variants (structured)
    title_variants = extract_title_variants(data)

    # Parent HMO URIs (part_of)
    parent_uris = extract_part_of(data)

    record_created, record_modified = extract_record_timestamps(data)
    attribution_evidence = extract_attribution_evidence(data)
    about_vocab_ids = extract_about(data)
    artwork_external_ids = extract_artwork_external_ids(data)
    visualitem_uri = extract_visualitem_uri(data)

    return {
        "inscription_text": inscription_text,
        "provenance_text": provenance_text,
        "credit_line": credit_line,
        "description_text": description_text,
        "height_cm": height_cm,
        "width_cm": width_cm,
        "depth_cm": depth_cm,
        "diameter_cm": diameter_cm,
        "weight_g": weight_g,
        "dimension_note": dimension_note,
        "narrative_text": narrative_text,
        "date_earliest": date_earliest,
        "date_latest": date_latest,
        "date_display": date_display,
        "current_location": current_location,
        "provenance_text_hash": provenance_text_hash,
        "title_all_text": title_all_text,
        "creator_label": creator_label,
        "roles": roles,
        "qualifiers": qualifiers,
        "creators": creators,
        "assignment_pairs": assignment_pairs,
        "places": places,
        "source_types": source_types,
        "modifications": modifications,
        "related_objects": related_objects,
        "examinations": examinations,
        "title_variants": title_variants,
        "parent_uris": parent_uris,
        "record_created": record_created,
        "record_modified": record_modified,
        "attribution_evidence": attribution_evidence,
        "about_vocab_ids": about_vocab_ids,
        "artwork_external_ids": artwork_external_ids,
        "visualitem_uri": visualitem_uri,
    }


def run_phase4(conn: sqlite3.Connection, threads: int = DEFAULT_THREADS):
    """Phase 4: Resolve all artwork Linked Art URIs for Tier 2 fields."""
    cur = conn.cursor()

    # Guard: Phase 3 drops tier2_done and linked_art_uri columns — Phase 4 cannot run after that
    artworks_cols = get_columns(conn, "artworks")
    if "tier2_done" not in artworks_cols or "linked_art_uri" not in artworks_cols:
        print("  tier2_done/linked_art_uri columns not present (Phase 3 already finalized).")
        print("  To re-run Tier 2 resolution, start from a fresh Phase 1+2 harvest.")
        return

    # Detect schema early — needed by seed step and MAPPING_INSERT_SQL below
    int_mappings = "field_id" in get_columns(conn, "mappings")

    # Seed AAT source-type vocabulary entries (used_object_of_type).
    # These are external AAT URIs not present in the Rijksmuseum vocabulary dump.
    # _extract_ids() emits the AAT suffix (e.g. "300102051") as vocab_id, so we insert
    # with that as the primary key to match the MAPPING_INSERT_SQL JOIN.
    # vocab_int_id is an explicit INTEGER column (NOT a rowid alias) used by the
    # integer-schema MAPPING_INSERT_SQL — seeded rows must get a value.
    max_vid = 0
    if int_mappings:
        max_vid = cur.execute("SELECT COALESCE(MAX(vocab_int_id), 0) FROM vocabulary").fetchone()[0]
    seeded = 0
    for aat_id, label_en in AAT_SOURCE_TYPES.items():
        res = cur.execute(
            "INSERT OR IGNORE INTO vocabulary (id, type, label_en) VALUES (?, 'classification', ?)",
            (aat_id, label_en),
        )
        if res.rowcount and int_mappings:
            max_vid += 1
            cur.execute("UPDATE vocabulary SET vocab_int_id = ? WHERE id = ?", (max_vid, aat_id))
        seeded += res.rowcount
    if seeded:
        if int_mappings:
            print(f"  Seeded {seeded} AAT source-type vocabulary entries (vocab_int_id {max_vid - seeded + 1}–{max_vid}).")
        else:
            print(f"  Seeded {seeded} AAT source-type vocabulary entries.")

    # Ensure new field names exist in field_lookup (integer-schema DBs).
    # The MAPPING_INSERT_SQL JOINs against field_lookup — missing entries cause silent drops.
    # 'theme' carries la-framed `about[]` thematic-vocab mappings.
    if int_mappings:
        fl_added = 0
        for field_name in ("production_place", "source_type", "theme"):
            res = cur.execute("INSERT OR IGNORE INTO field_lookup (name) VALUES (?)", (field_name,))
            fl_added += res.rowcount
        if fl_added:
            print(f"  Added {fl_added} field_lookup row(s).")

    # Ensure art_id exists on artworks before Phase 4's main loop — the six
    # new-table inserts (modifications, related_objects, examinations,
    # title_variants, assignment_pairs, artwork_parent) are keyed by art_id
    # and would otherwise be silently skipped on a fresh harvest because
    # Phase 3's normalize_mappings() doesn't run until AFTER Phase 4.
    #
    # This block is idempotent — normalize_mappings() has the same guard and
    # will skip the art_id bootstrap if it's already in place. Fix for #219
    # (six Phase 4 tables returned 0 rows in v0.24).
    if "art_id" not in get_columns(conn, "artworks"):
        print("  Adding stable art_id column (so Phase 4 new-table inserts work)...")
        conn.execute("ALTER TABLE artworks ADD COLUMN art_id INTEGER")
        conn.execute("UPDATE artworks SET art_id = rowid")
        conn.execute("CREATE UNIQUE INDEX idx_artworks_art_id ON artworks(art_id)")

    conn.commit()

    # Get all artworks that haven't been processed yet
    pending = cur.execute("""
        SELECT object_number, linked_art_uri
        FROM artworks
        WHERE tier2_done = 0 AND linked_art_uri IS NOT NULL AND linked_art_uri != ''
    """).fetchall()

    if not pending:
        print("  No artworks pending Tier 2 resolution.")
        return

    total = len(pending)
    print(f"  Resolving {total:,} artworks for Tier 2 fields ({threads} threads)...")

    # Detect schema for mappings inserts (production roles + attribution qualifiers).
    # art_id is guaranteed present by the bootstrap block above — but keep the
    # dynamic check so this function stays robust against schema drift.
    artworks_cols = get_columns(conn, "artworks")
    has_art_id = "art_id" in artworks_cols

    # Pre-fetch {object_number → art_id} so the per-artwork loop can resolve
    # art_id via dict lookup instead of N x "SELECT art_id FROM artworks WHERE
    # object_number = ?". At 833K artworks, the pre-fetch saves ~833K SQLite
    # round-trips (~minute+ of wall-clock time). Costs one full-table scan
    # and ~60 MB of process memory.
    art_id_map: dict[str, int] = {}
    if has_art_id:
        art_id_map = dict(conn.execute("SELECT object_number, art_id FROM artworks").fetchall())

    MAPPING_INSERT_SQL = _mapping_insert_sql(int_mappings)

    processed = 0
    succeeded = 0
    failed = 0
    not_found = 0
    with_inscription = 0
    with_provenance = 0
    with_credit = 0
    with_description = 0
    with_dimensions = 0
    with_narrative = 0
    with_dates = 0
    with_titles = 0
    with_creator_label = 0
    role_count = 0
    qualifier_count = 0
    creator_count = 0
    modification_count = 0
    related_object_count = 0
    examination_count = 0
    title_variant_count = 0
    assignment_pair_count = 0
    parent_count = 0
    place_count = 0
    source_type_count = 0
    attribution_evidence_count = 0
    about_count = 0
    artwork_external_id_count = 0
    with_record_timestamp = 0
    t0 = time.time()

    TIER2_UPDATE_SQL = """
        UPDATE artworks SET
            inscription_text = ?,
            provenance_text = ?,
            credit_line = ?,
            description_text = ?,
            height_cm = ?,
            width_cm = ?,
            depth_cm = ?,
            diameter_cm = ?,
            weight_g = ?,
            dimension_note = ?,
            narrative_text = ?,
            date_earliest = ?,
            date_latest = ?,
            date_display = ?,
            current_location = ?,
            provenance_text_hash = ?,
            title_all_text = ?,
            creator_label = COALESCE(?, creator_label),
            record_created = ?,
            record_modified = ?,
            visualitem_uri = ?,
            tier2_done = 1
        WHERE object_number = ?
    """

    with ThreadPoolExecutor(max_workers=threads) as pool:
        # Submit in batches to avoid excessive memory for futures dict
        batch_start = 0
        commit_batch_size = 500

        while batch_start < total:
            batch_end = min(batch_start + commit_batch_size, total)
            batch = pending[batch_start:batch_end]

            futures = {
                pool.submit(resolve_artwork, uri): (obj_num, uri)
                for obj_num, uri in batch
            }

            for future in as_completed(futures):
                obj_num, uri = futures[future]
                processed += 1

                try:
                    result = future.result()
                except Exception:
                    result = None

                if result is None:
                    failed += 1
                    # Don't mark tier2_done — leave for retry
                    continue

                if result.get("_status") == "not_found":
                    not_found += 1
                    # Mark as done so we don't retry 404s
                    conn.execute("UPDATE artworks SET tier2_done = 1 WHERE object_number = ?", (obj_num,))
                    continue

                succeeded += 1
                if result["inscription_text"]:
                    with_inscription += 1
                if result["provenance_text"]:
                    with_provenance += 1
                if result["credit_line"]:
                    with_credit += 1
                if result["description_text"]:
                    with_description += 1
                if result["height_cm"] is not None or result["width_cm"] is not None:
                    with_dimensions += 1
                if result["narrative_text"]:
                    with_narrative += 1
                if result["date_earliest"] is not None:
                    with_dates += 1
                if result["title_all_text"]:
                    with_titles += 1
                if result.get("creator_label"):
                    with_creator_label += 1

                conn.execute(TIER2_UPDATE_SQL, (
                    result["inscription_text"],
                    result["provenance_text"],
                    result["credit_line"],
                    result["description_text"],
                    result["height_cm"],
                    result["width_cm"],
                    result["depth_cm"],
                    result["diameter_cm"],
                    result["weight_g"],
                    result["dimension_note"],
                    result["narrative_text"],
                    result["date_earliest"],
                    result["date_latest"],
                    result["date_display"],
                    result["current_location"],
                    result["provenance_text_hash"],
                    result["title_all_text"],
                    result.get("creator_label"),
                    result.get("record_created"),
                    result.get("record_modified"),
                    result.get("visualitem_uri"),
                    obj_num,
                ))
                if result.get("record_created") or result.get("record_modified"):
                    with_record_timestamp += 1

                # Insert production role, attribution qualifier, creator, place, and source type mappings
                for vocab_id, field in chain(result["roles"], result["qualifiers"], result["creators"], result["places"], result["source_types"]):
                    conn.execute(MAPPING_INSERT_SQL, (obj_num, vocab_id, field))
                role_count += len(result["roles"])
                qualifier_count += len(result["qualifiers"])
                creator_count += len(result["creators"])
                place_count += len(result["places"])
                source_type_count += len(result["source_types"])

                # New-table inserts require art_id. The bootstrap block at the top
                # of run_phase4 guarantees art_id exists, so has_art_id should be
                # True — but keep the fallback so this function stays robust if
                # the bootstrap is ever disabled or the caller reshapes the schema.
                art_id = art_id_map.get(obj_num) if has_art_id else None

                if art_id is not None:
                    # Insert modifications
                    for seq, mod in enumerate(result.get("modifications", [])):
                        conn.execute(
                            "INSERT OR IGNORE INTO modifications (art_id, seq, modifier_uri, date_display, date_begin, date_end, description) VALUES (?, ?, ?, ?, ?, ?, ?)",
                            (art_id, seq, mod["modifier_uri"], mod["date_display"], mod["date_begin"], mod["date_end"], mod["description"]),
                        )
                    modification_count += len(result.get("modifications", []))

                    # Insert related objects
                    for ro in result.get("related_objects", []):
                        conn.execute(
                            "INSERT OR IGNORE INTO related_objects (art_id, related_la_uri, relationship_en, relationship_nl) VALUES (?, ?, ?, ?)",
                            (art_id, ro["related_la_uri"], ro["relationship_en"], ro.get("relationship_nl")),
                        )
                    related_object_count += len(result.get("related_objects", []))

                    # Insert examinations
                    for seq, exam in enumerate(result.get("examinations", [])):
                        conn.execute(
                            "INSERT OR IGNORE INTO examinations (art_id, seq, examiner_name, report_type_id, report_type_en, date_display, date_begin, date_end) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                            (art_id, seq, exam["examiner_name"], exam["report_type_id"], exam.get("report_type_en"), exam.get("date_display"), exam.get("date_begin"), exam.get("date_end")),
                        )
                    examination_count += len(result.get("examinations", []))

                    # Insert title variants
                    for seq, tv in enumerate(result.get("title_variants", [])):
                        conn.execute(
                            "INSERT OR IGNORE INTO title_variants (art_id, seq, title_text, language, qualifier) VALUES (?, ?, ?, ?, ?)",
                            (art_id, seq, tv["title_text"], tv.get("language"), tv.get("qualifier")),
                        )
                    title_variant_count += len(result.get("title_variants", []))

                    # Insert assignment pairs
                    for q_id, c_id, part_idx in result.get("assignment_pairs", []):
                        conn.execute(
                            "INSERT OR IGNORE INTO assignment_pairs (artwork_id, qualifier_id, creator_id, part_index) VALUES (?, ?, ?, ?)",
                            (art_id, q_id, c_id, part_idx),
                        )
                    assignment_pair_count += len(result.get("assignment_pairs", []))

                    # Insert parent URIs
                    for parent_uri in result.get("parent_uris", []):
                        conn.execute(
                            "INSERT OR IGNORE INTO artwork_parent (art_id, parent_la_uri) VALUES (?, ?)",
                            (art_id, parent_uri),
                        )
                    parent_count += len(result.get("parent_uris", []))

                    evidence = result.get("attribution_evidence")
                    if evidence:
                        conn.executemany(
                            "INSERT OR IGNORE INTO attribution_evidence "
                            "(art_id, part_index, evidence_type_aat, carried_by_uri, label_text) "
                            "VALUES (?, ?, ?, ?, ?)",
                            [
                                (art_id, ev["part_index"], ev["evidence_type_aat"],
                                 ev["carried_by_uri"], ev["label_text"])
                                for ev in evidence
                            ],
                        )
                        attribution_evidence_count += len(evidence)

                    ext_ids = result.get("artwork_external_ids")
                    if ext_ids:
                        conn.executemany(
                            "INSERT OR IGNORE INTO artwork_external_ids "
                            "(art_id, authority, id, uri) VALUES (?, ?, ?, ?)",
                            [(art_id, auth, lid, uri) for auth, lid, uri in ext_ids],
                        )
                        artwork_external_id_count += len(ext_ids)

                # Thematic vocab mappings (field='theme'). Uses MAPPING_INSERT_SQL
                # like other vocab fields; orphan vocab_ids (entity not yet
                # harvested) are silently dropped during integer encoding.
                about_ids = result.get("about_vocab_ids")
                if about_ids:
                    conn.executemany(
                        MAPPING_INSERT_SQL,
                        [(obj_num, vid, "theme") for vid in about_ids],
                    )
                    about_count += len(about_ids)

            conn.commit()
            batch_start = batch_end

            elapsed = time.time() - t0
            rate = processed / elapsed if elapsed > 0 else 0
            remaining = (total - processed) / rate if rate > 0 else 0
            print(
                f"  {processed:,}/{total:,} ({succeeded:,} ok, {failed:,} failed, {not_found:,} 404, "
                f"{rate:.0f}/s, ~{remaining / 60:.0f}min left)",
                flush=True,
            )

    elapsed = time.time() - t0
    print(f"\n  Phase 4 complete in {elapsed / 60:.1f}min:")
    print(f"    Succeeded:    {succeeded:,}")
    print(f"    Failed:       {failed:,} (will retry on --resume --phase 4)")
    print(f"    Not found:    {not_found:,}")
    print(f"    Inscriptions: {with_inscription:,}")
    print(f"    Provenance:   {with_provenance:,}")
    print(f"    Credit lines: {with_credit:,}")
    print(f"    Descriptions: {with_description:,}")
    print(f"    Dimensions:   {with_dimensions:,}")
    print(f"    Narratives:   {with_narrative:,}")
    print(f"    Dates:        {with_dates:,}")
    print(f"    All titles:   {with_titles:,}")
    print(f"    Creator label:{with_creator_label:,}")
    print(f"    Prod. roles:  {role_count:,}")
    print(f"    Attr. quals:  {qualifier_count:,}")
    print(f"    Creators:     {creator_count:,} (from assigned_by)")
    print(f"    Modifications:{modification_count:,}")
    print(f"    Related obj:  {related_object_count:,}")
    print(f"    Examinations: {examination_count:,}")
    print(f"    Title vars:   {title_variant_count:,}")
    print(f"    Assign. pairs:{assignment_pair_count:,}")
    print(f"    Parent links: {parent_count:,}")
    print(f"    Prod. places: {place_count:,}")
    print(f"    Source types: {source_type_count:,}")
    print(f"    Record TS:    {with_record_timestamp:,}")
    print(f"    Attr. evidence:{attribution_evidence_count:,}")
    print(f"    about[] themes:{about_count:,}")
    print(f"    Artwork ext IDs:{artwork_external_id_count:,}")


# ─── Phase 4.5: VisualItem.about[] dereference (#283) ────────────────

def _fetch_visualitem_about(uri: str) -> tuple[str, list[str] | None]:
    """Fetch a VisualItem URI and return (uri, about_vocab_ids).

    On HTTP/JSON failure returns (uri, None) so the caller can count it
    against the failure tally without aborting the rest of the pass.
    """
    try:
        resp = get_http_session().get(uri, headers={
            "Accept": "application/ld+json",
            "Profile": "https://linked.art/ns/v1/linked-art.json",
            "User-Agent": USER_AGENT,
        }, timeout=30)
    except requests.RequestException:
        return uri, None
    if resp.status_code == 404:
        return uri, []  # no VI: real not-found, treat as zero-themes
    if not resp.ok:
        return uri, None
    try:
        data = resp.json()
    except ValueError:
        return uri, None
    return uri, extract_about(data)


def run_phase4_5(conn: sqlite3.Connection, threads: int = DEFAULT_THREADS):
    """Phase 4.5: dereference each HMO's VisualItem to capture `about[]`
    thematic-vocab links. The data-path correction here is non-obvious
    (about[] lives on VisualItem, not HMO).

    Reads the visualitem_uri column populated during Phase 4. Idempotent:
    re-running on a populated DB is a no-op via INSERT OR IGNORE on the
    mappings table.
    """
    cur = conn.cursor()

    artworks_cols = get_columns(conn, "artworks")
    if "visualitem_uri" not in artworks_cols:
        print("  visualitem_uri column dropped — Phase 3 finalization already ran.")
        return

    rows = cur.execute(
        "SELECT object_number, visualitem_uri FROM artworks "
        "WHERE visualitem_uri IS NOT NULL AND visualitem_uri != ''"
    ).fetchall()
    total = len(rows)
    if total == 0:
        print("  No visualitem_uri values to fetch — Phase 4 may not have run.")
        return

    int_mappings = "field_id" in get_columns(conn, "mappings")
    insert_sql = _mapping_insert_sql(int_mappings)

    print(f"  Resolving {total:,} VisualItems for thematic vocab ({threads} threads)...")
    t0 = time.time()
    processed = 0
    failed = 0
    no_themes = 0
    with_themes = 0
    total_theme_rows = 0

    batch_size = 500
    pending: list[tuple[str, str, str]] = []

    with ThreadPoolExecutor(max_workers=threads) as pool:
        for batch_start in range(0, total, batch_size):
            batch = rows[batch_start:batch_start + batch_size]
            futures = {pool.submit(_fetch_visualitem_about, vi): obj for obj, vi in batch}
            for future in as_completed(futures):
                obj_num = futures[future]
                processed += 1
                try:
                    _, about_ids = future.result()
                except Exception:
                    failed += 1
                    continue
                if about_ids is None:
                    failed += 1
                    continue
                if not about_ids:
                    no_themes += 1
                    continue
                with_themes += 1
                for vocab_id in about_ids:
                    pending.append((obj_num, vocab_id, "theme"))
                total_theme_rows += len(about_ids)

            conn.executemany(insert_sql, pending)
            conn.commit()
            pending.clear()

            if (batch_start // batch_size) % 20 == 0 and processed > 0:
                elapsed = time.time() - t0
                rate = processed / elapsed if elapsed > 0 else 0
                remaining = (total - processed) / rate if rate > 0 else 0
                print(
                    f"    {processed:,}/{total:,} ({with_themes:,} with themes, "
                    f"{failed:,} failed, {no_themes:,} no-themes, "
                    f"{rate:.0f}/s, ~{remaining / 60:.0f}min left)",
                    flush=True,
                )

    elapsed = time.time() - t0
    print(f"\n  Phase 4.5 complete in {elapsed / 60:.1f}min:")
    print(f"    Processed:    {processed:,}")
    print(f"    With themes:  {with_themes:,}")
    print(f"    No-themes:    {no_themes:,}")
    print(f"    Failed:       {failed:,}")
    print(f"    Theme rows:   {total_theme_rows:,}")


# ─── Geocoding Import ────────────────────────────────────────────────

def import_geocoding(conn: sqlite3.Connection, csv_path: str):
    """Import geocoded place data from CSV into the vocabulary table.

    Updates lat, lon, external_id, and (when present) the v0.25 provenance
    columns ``coord_method`` / ``coord_method_detail`` /
    ``external_id_method`` / ``external_id_method_detail`` /
    ``broader_method`` / ``broader_method_detail``.

    The CSV may use either ``id`` or ``vocab_id`` as the row-key column.
    Provenance columns are optional — older CSVs without them are imported
    with NULLs in those columns (back-compat).
    """
    csv_file = Path(csv_path)
    if not csv_file.exists():
        print(f"  ERROR: Geocoding CSV not found: {csv_path}")
        return

    print(f"  Reading: {csv_path}")
    updated_coords = 0
    updated_ext_id = 0
    updated_method = 0
    skipped_missing = 0
    skipped_not_place = 0

    METHOD_COLS = (
        "coord_method", "coord_method_detail",
        "external_id_method", "external_id_method_detail",
        "broader_method", "broader_method_detail",
    )

    with open(csv_file, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        has_method_cols = bool(reader.fieldnames) and any(
            c in reader.fieldnames for c in METHOD_COLS
        )
        batch = []

        for row in reader:
            vocab_id = (row.get("id") or row.get("vocab_id") or "").strip()
            lat_str = (row.get("lat") or "").strip()
            lon_str = (row.get("lon") or "").strip()
            ext_id = (row.get("external_id") or "").strip()

            if not vocab_id or not lat_str or not lon_str:
                continue

            try:
                lat = float(lat_str)
                lon = float(lon_str)
            except ValueError:
                continue

            method_values = tuple(
                ((row.get(c) or "").strip() or None) for c in METHOD_COLS
            ) if has_method_cols else (None, None, None, None, None, None)

            batch.append((vocab_id, lat, lon, ext_id, method_values))

        print(f"  Parsed {len(batch):,} geocoded places from CSV "
              f"(method cols: {'yes' if has_method_cols else 'no'})")

    # Bulk-load existing rows for the CSV's vocab IDs. Single SELECT vs.
    # one-per-row avoids ~N round-trips on a 30k-row import.
    batch_ids = list({row[0] for row in batch})
    existing_map: dict[str, tuple] = {}
    chunk = 900  # SQLite default SQLITE_MAX_VARIABLE_NUMBER is 999
    for i in range(0, len(batch_ids), chunk):
        slice_ = batch_ids[i:i + chunk]
        placeholders = ",".join("?" * len(slice_))
        rows = conn.execute(
            f"SELECT id, type, external_id FROM vocabulary WHERE id IN ({placeholders})",
            slice_,
        ).fetchall()
        for r in rows:
            existing_map[r[0]] = (r[1], r[2])

    for vocab_id, lat, lon, ext_id, method_values in batch:
        existing = existing_map.get(vocab_id)
        if existing is None:
            skipped_missing += 1
            continue

        if existing[0] != "place":
            skipped_not_place += 1
            continue

        should_update_ext = ext_id and ext_id != existing[1]
        if should_update_ext:
            conn.execute(
                "UPDATE vocabulary "
                "   SET lat = ?, lon = ?, "
                "       geocode_method = 'csv', external_id = ?, "
                "       coord_method = ?, coord_method_detail = ?, "
                "       external_id_method = ?, external_id_method_detail = ?, "
                "       broader_method = ?, broader_method_detail = ? "
                " WHERE id = ?",
                (lat, lon, ext_id, *method_values, vocab_id),
            )
            updated_ext_id += 1
        else:
            conn.execute(
                "UPDATE vocabulary "
                "   SET lat = ?, lon = ?, "
                "       geocode_method = 'csv', "
                "       coord_method = ?, coord_method_detail = ?, "
                "       external_id_method = ?, external_id_method_detail = ?, "
                "       broader_method = ?, broader_method_detail = ? "
                " WHERE id = ?",
                (lat, lon, *method_values, vocab_id),
            )
        updated_coords += 1
        if has_method_cols and any(method_values):
            updated_method += 1

    conn.commit()
    print(f"  Geocoding import complete:")
    print(f"    Coordinates updated: {updated_coords:,}")
    print(f"    External IDs updated: {updated_ext_id:,}")
    print(f"    Provenance tags written: {updated_method:,}")
    if skipped_missing:
        print(f"    Skipped (not in DB): {skipped_missing:,}")
    if skipped_not_place:
        print(f"    Skipped (not a place): {skipped_not_place:,}")


# ─── Normalization (integer-encoding) ─────────────────────────────────

def normalize_mappings(conn: sqlite3.Connection):
    """Integer-encode the mappings table for ~1.2 GB space savings.

    Replaces TEXT columns (object_number, vocab_id, field) with INTEGER FKs
    (artwork_id, vocab_rowid, field_id) in a WITHOUT ROWID clustered B-tree.
    Idempotent: skips if already normalized.
    """
    # Guard: skip if already normalized
    if "artwork_id" in get_columns(conn, "mappings"):
        print("  Mappings already integer-encoded, skipping.")
        return

    t0 = time.time()
    cur = conn.cursor()

    # Step 1: Add stable integer IDs to parent tables (survive VACUUM)
    print("  Adding stable integer IDs to artworks and vocabulary...")
    if "art_id" not in get_columns(conn, "artworks"):
        conn.execute("ALTER TABLE artworks ADD COLUMN art_id INTEGER")
        conn.execute("UPDATE artworks SET art_id = rowid")
        conn.execute("CREATE UNIQUE INDEX idx_artworks_art_id ON artworks(art_id)")

    if "vocab_int_id" not in get_columns(conn, "vocabulary"):
        conn.execute("ALTER TABLE vocabulary ADD COLUMN vocab_int_id INTEGER")
        conn.execute("UPDATE vocabulary SET vocab_int_id = rowid")
        conn.execute("CREATE UNIQUE INDEX idx_vocab_int_id ON vocabulary(vocab_int_id)")
    conn.commit()

    # Step 2: Create field_lookup table
    print("  Creating field_lookup table...")
    conn.execute("DROP TABLE IF EXISTS field_lookup")
    conn.execute("CREATE TABLE field_lookup (id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE)")
    conn.execute("INSERT INTO field_lookup (name) SELECT DISTINCT field FROM mappings ORDER BY field")
    field_count = cur.execute("SELECT COUNT(*) FROM field_lookup").fetchone()[0]
    print(f"    {field_count} distinct fields")
    conn.commit()

    # Step 3: Create and populate integer-encoded mappings table
    #   WITHOUT ROWID stores the composite INTEGER PK as a clustered B-tree,
    #   avoiding separate rowid allocation and saving ~50% vs regular table.
    print("  Building integer-encoded mappings table (this may take a few minutes)...")
    conn.execute("DROP TABLE IF EXISTS mappings_int")
    conn.execute("""
        CREATE TABLE mappings_int (
            artwork_id  INTEGER NOT NULL,
            vocab_rowid INTEGER NOT NULL,
            field_id    INTEGER NOT NULL,
            PRIMARY KEY (artwork_id, vocab_rowid, field_id)
        ) WITHOUT ROWID
    """)
    conn.execute("""
        INSERT INTO mappings_int (artwork_id, vocab_rowid, field_id)
        SELECT a.art_id, v.vocab_int_id, f.id
        FROM mappings m
        JOIN artworks a ON m.object_number = a.object_number
        JOIN vocabulary v ON m.vocab_id = v.id
        JOIN field_lookup f ON m.field = f.name
    """)
    new_count = cur.execute("SELECT COUNT(*) FROM mappings_int").fetchone()[0]
    old_count = cur.execute("SELECT COUNT(*) FROM mappings").fetchone()[0]
    print(f"    Migrated {new_count:,} of {old_count:,} mappings")
    if new_count < old_count:
        print(f"    Note: {old_count - new_count:,} orphaned mappings dropped (vocab_id not in vocabulary)")
    conn.commit()

    # Step 4: Swap tables (rename-then-drop is crash-safe: if killed after
    # the first rename, both tables survive and the idempotency guard recovers)
    print("  Swapping tables...")
    conn.execute("ALTER TABLE mappings RENAME TO mappings_old")
    conn.execute("ALTER TABLE mappings_int RENAME TO mappings")
    conn.commit()
    conn.execute("DROP TABLE mappings_old")
    conn.commit()

    # Step 5: Recreate secondary indexes
    # Only idx_mappings_field_vocab is needed. idx_mappings_field_artwork is a
    # performance anti-pattern (9,000-17,000x slower on enrichment queries) and
    # idx_mappings_vocab is redundant (not used by any runtime query path).
    print("  Creating secondary indexes...")
    conn.execute("CREATE INDEX idx_mappings_field_vocab ON mappings(field_id, vocab_rowid)")
    conn.commit()

    elapsed = time.time() - t0
    print(f"  Integer-encoding complete in {elapsed:.1f}s")


def normalize_rights(conn: sqlite3.Connection):
    """Normalize artworks.rights_uri to integer FK via rights_lookup table.

    Creates a rights_lookup table (3-4 rows) and replaces the TEXT rights_uri
    column with an INTEGER rights_id FK. Idempotent: skips if already normalized.
    """
    # Guard: skip if already normalized
    if "rights_id" in get_columns(conn, "artworks"):
        print("  Rights already normalized, skipping.")
        return

    print("  Normalizing rights_uri to integer FK...")

    conn.execute("DROP TABLE IF EXISTS rights_lookup")
    conn.execute("CREATE TABLE rights_lookup (id INTEGER PRIMARY KEY, uri TEXT NOT NULL UNIQUE)")
    conn.execute("""
        INSERT INTO rights_lookup (uri)
        SELECT DISTINCT rights_uri FROM artworks
        WHERE rights_uri IS NOT NULL AND rights_uri != ''
    """)
    rights_count = conn.execute("SELECT COUNT(*) FROM rights_lookup").fetchone()[0]
    print(f"    {rights_count} distinct rights URIs")

    conn.execute("ALTER TABLE artworks ADD COLUMN rights_id INTEGER")
    conn.execute("""
        UPDATE artworks SET rights_id = (
            SELECT r.id FROM rights_lookup r WHERE r.uri = artworks.rights_uri
        )
    """)
    conn.commit()

    # Drop the old TEXT column (SQLite >= 3.35.0)
    try:
        conn.execute("ALTER TABLE artworks DROP COLUMN rights_uri")
        conn.commit()
        print("    Dropped rights_uri column")
    except Exception:
        print("    Note: Could not drop rights_uri (SQLite < 3.35.0), column retained")


# ─── Enrichment (folded in from enrich-vocab-from-dumps.py, #242 part 3) ───
#
# These functions enrich the vocabulary table from the four offline data dumps
# in DUMPS_DIR. They run inside run_phase3() after integer encoding, and are
# idempotent (every UPDATE is gated on `AND <col> IS NULL`). When dump artefacts
# are missing, each phase logs a warning and returns; the Part-1 column stubs
# ensure the server can still open the DB.

_ENRICH_NS = {
    "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
    "skos": "http://www.w3.org/2004/02/skos/core#",
    "dc": "http://purl.org/dc/elements/1.1/",
    "rdaGr2": "http://rdvocab.info/ElementsGr2/",
    "edm": "http://www.europeana.eu/schemas/edm/",
}

_ENRICH_RE_FALLS_WITHIN = re.compile(
    r"<https://id\.rijksmuseum\.nl/(\d+)>\s+"
    r"<http://www\.cidoc-crm\.org/cidoc-crm/P89_falls_within>\s+"
    r"<https://id\.rijksmuseum\.nl/(\d+)>"
)
_ENRICH_RE_EQUIVALENT = re.compile(
    r"<https://id\.rijksmuseum\.nl/(\d+)>\s+"
    r"<https://linked\.art/ns/terms/equivalent>\s+"
    r"<(http[^>]+)>"
)
_ENRICH_RE_SCHEMA_NAME = re.compile(
    r"<https://id\.rijksmuseum\.nl/(\d+)>\s+"
    r"<http://schema\.org/name>\s+"
    r'"([^"]*)"'
)
_ENRICH_RE_SCHEMA_ALT_NAME = re.compile(
    r"<https://id\.rijksmuseum\.nl/(\d+)>\s+"
    r"<http://schema\.org/alternateName>\s+"
    r'"([^"]*)"'
)
# Combined pattern that matches both schema:name and schema:alternateName in
# a single scan — halves the regex passes per file in the dump enrichment.
_ENRICH_RE_SCHEMA_NAME_OR_ALT = re.compile(
    r"<https://id\.rijksmuseum\.nl/(\d+)>\s+"
    r"<http://schema\.org/(name|alternateName)>\s+"
    r'"([^"]*)"'
)
_ENRICH_RE_SCHEMA_SAME_AS = re.compile(
    r"<https://id\.rijksmuseum\.nl/(\d+)>\s+"
    r"<http://schema\.org/sameAs>\s+"
    r"<(http[^>]+)>"
)
_ENRICH_RE_YEAR = re.compile(r"\b(\d{4})\b")
_ENRICH_RE_WIKIDATA_QID = re.compile(r"wikidata\.org/entity/(Q\d+)")


def _enrich_parse_year(text):
    if not text:
        return None
    m = _ENRICH_RE_YEAR.search(text)
    return int(m.group(1)) if m else None


def enrich_places_from_dump(conn: sqlite3.Connection, dumps_dir: Path) -> None:
    """Enrich places with broader_id and external_id from the 2025 place dump.

    Matches dump files by filename (filename IS the place id) against
    vocabulary entries of type='place'. Prefers Wikidata > TGN for external_id.
    """
    place_dir = dumps_dir / ENRICH_PLACE_DIR
    if not place_dir.exists():
        print(f"  [enrich places] {place_dir} not found — skipping.")
        return

    vocab_place_ids = {
        r[0] for r in conn.execute(
            "SELECT id FROM vocabulary WHERE type='place'"
        ).fetchall()
    }
    print(f"  [enrich places] Vocab places: {len(vocab_place_ids):,}")

    files = os.listdir(place_dir)
    print(f"  [enrich places] Place dump files: {len(files):,}")

    batch = []
    skipped_not_in_vocab = 0
    for i, fname in enumerate(files):
        if i % 5000 == 0 and i > 0:
            print(f"    ... processed {i:,}/{len(files):,}")
        fpath = place_dir / fname
        try:
            text = fpath.read_text()
        except Exception:
            continue
        place_id = fname
        if place_id not in vocab_place_ids:
            skipped_not_in_vocab += 1
            continue

        parent_id = None
        for m in _ENRICH_RE_FALLS_WITHIN.finditer(text):
            if m.group(1) == place_id:
                parent_id = m.group(2)
                break

        external_uris = []
        for m in _ENRICH_RE_EQUIVALENT.finditer(text):
            if m.group(1) == place_id:
                uri = m.group(2)
                if not uri.startswith("https://id.rijksmuseum.nl/330"):
                    external_uris.append(uri)
        best_external = None
        for uri in external_uris:
            if "wikidata.org" in uri:
                best_external = uri
                break
        if not best_external:
            for uri in external_uris:
                if "getty.edu/tgn" in uri:
                    best_external = uri
                    break

        batch.append((place_id, parent_id, best_external))

    updates_broader = 0
    updates_external = 0
    for place_id, parent_id, best_external in batch:
        if parent_id:
            cur = conn.execute(
                "UPDATE vocabulary SET broader_id = ? WHERE id = ? AND broader_id IS NULL",
                (parent_id, place_id),
            )
            updates_broader += cur.rowcount
        if best_external:
            cur = conn.execute(
                "UPDATE vocabulary SET external_id = ? WHERE id = ? AND external_id IS NULL",
                (best_external, place_id),
            )
            updates_external += cur.rowcount
    conn.commit()
    print(
        f"  [enrich places] Updated {updates_broader:,} broader_id, "
        f"{updates_external:,} external_id "
        f"(dump IDs not in vocab: {skipped_not_in_vocab:,})"
    )


def enrich_actors_from_edm(conn: sqlite3.Connection, dumps_dir: Path) -> None:
    """Enrich persons with birth_year/death_year/gender from 2019 EDM actors dump.

    Matches by prefLabel then altLabels against person_names. When multiple
    dump entries hit the same vocab id, keeps the richest (most non-NULL fields).
    """
    zip_path = dumps_dir / ENRICH_EDM_ACTORS_ZIP
    if not zip_path.exists():
        print(f"  [enrich actors] {zip_path} not found — skipping.")
        return

    # person_names table must exist; early harvest runs may not have it yet.
    try:
        conn.execute("SELECT 1 FROM person_names LIMIT 1").fetchone()
    except sqlite3.OperationalError:
        print("  [enrich actors] person_names table not present — skipping.")
        return

    print("  [enrich actors] Building name lookup...")
    name_to_ids = defaultdict(set)
    for person_id, name in conn.execute("SELECT person_id, name FROM person_names"):
        if name:
            name_to_ids[name].add(person_id)
    print(
        f"  [enrich actors] Name lookup: {len(name_to_ids):,} names → "
        f"{len(set().union(*name_to_ids.values())) if name_to_ids else 0:,} persons"
    )

    actors = []
    with zipfile.ZipFile(zip_path) as z:
        with z.open(z.namelist()[0]) as f:
            context = ET.iterparse(f, events=("end",))
            for _event, elem in context:
                if elem.tag == f"{{{_ENRICH_NS['edm']}}}Agent":
                    pref_label = None
                    alt_labels = []
                    birth = None
                    death = None
                    gender = None
                    for child in elem:
                        tag = child.tag
                        if tag == f"{{{_ENRICH_NS['skos']}}}prefLabel":
                            pref_label = child.text
                        elif tag == f"{{{_ENRICH_NS['skos']}}}altLabel":
                            if child.text:
                                alt_labels.append(child.text)
                        elif tag == f"{{{_ENRICH_NS['rdaGr2']}}}dateOfBirth":
                            birth = _enrich_parse_year(child.text)
                        elif tag == f"{{{_ENRICH_NS['rdaGr2']}}}dateOfDeath":
                            death = _enrich_parse_year(child.text)
                        elif tag == f"{{{_ENRICH_NS['rdaGr2']}}}gender":
                            gender = child.text
                    if pref_label:
                        richness = sum(
                            x is not None for x in (birth, death, gender)
                        )
                        actors.append(
                            (pref_label, alt_labels, birth, death, gender, richness)
                        )
                    elem.clear()
    print(f"  [enrich actors] Parsed {len(actors):,} EDM agents")

    vocab_id_to_data = {}
    matched = 0
    for pref_label, alt_labels, birth, death, gender, richness in actors:
        candidate_ids = name_to_ids.get(pref_label, set())
        if not candidate_ids:
            for alt in alt_labels:
                candidate_ids = name_to_ids.get(alt, set())
                if candidate_ids:
                    break
        if not candidate_ids:
            continue
        matched += 1
        for vid in candidate_ids:
            existing = vocab_id_to_data.get(vid)
            if existing is None or richness > existing[3]:
                vocab_id_to_data[vid] = (birth, death, gender, richness)
    print(
        f"  [enrich actors] Matched {matched:,} dump entries → "
        f"{len(vocab_id_to_data):,} unique vocab ids"
    )

    updates = {"birth_year": 0, "death_year": 0, "gender": 0}
    for vid, (birth, death, gender, _) in vocab_id_to_data.items():
        if birth is not None:
            updates["birth_year"] += conn.execute(
                "UPDATE vocabulary SET birth_year = ? WHERE id = ? AND birth_year IS NULL",
                (birth, vid),
            ).rowcount
        if death is not None:
            updates["death_year"] += conn.execute(
                "UPDATE vocabulary SET death_year = ? WHERE id = ? AND death_year IS NULL",
                (death, vid),
            ).rowcount
        if gender is not None:
            updates["gender"] += conn.execute(
                "UPDATE vocabulary SET gender = ? WHERE id = ? AND gender IS NULL",
                (gender, vid),
            ).rowcount
    conn.commit()
    print(f"  [enrich actors] Updates: {updates}")


def enrich_wikidata_from_person_dump(conn: sqlite3.Connection, dumps_dir: Path) -> None:
    """Enrich persons with wikidata_id from the 2025 person dump.

    Tries direct filename-ID match first, then falls back to schema:name /
    schema:alternateName lookups against person_names.
    """
    person_dir = dumps_dir / ENRICH_PERSON_DIR
    if not person_dir.exists():
        print(f"  [enrich wikidata] {person_dir} not found — skipping.")
        return

    try:
        conn.execute("SELECT 1 FROM person_names LIMIT 1").fetchone()
    except sqlite3.OperationalError:
        print("  [enrich wikidata] person_names table not present — skipping.")
        return

    vocab_person_ids = {
        r[0] for r in conn.execute(
            "SELECT id FROM vocabulary WHERE type='person'"
        ).fetchall()
    }
    name_to_ids = defaultdict(set)
    for person_id, name in conn.execute("SELECT person_id, name FROM person_names"):
        if name:
            name_to_ids[name].add(person_id)
    files = os.listdir(person_dir)
    print(
        f"  [enrich wikidata] Vocab persons: {len(vocab_person_ids):,}, "
        f"dump files: {len(files):,}"
    )

    wikidata_updates = {}
    direct_matches = 0
    name_matches = 0
    for i, fname in enumerate(files):
        if i % 20000 == 0 and i > 0:
            print(f"    ... processed {i:,}/{len(files):,}")
        fpath = person_dir / fname
        try:
            text = fpath.read_text()
        except Exception:
            continue
        dump_id = fname

        wikidata_qid = None
        for m in _ENRICH_RE_SCHEMA_SAME_AS.finditer(text):
            if m.group(1) == dump_id:
                uri = m.group(2)
                qm = _ENRICH_RE_WIKIDATA_QID.search(uri)
                if qm:
                    wikidata_qid = qm.group(1)
                    break
        if not wikidata_qid:
            continue

        if dump_id in vocab_person_ids:
            wikidata_updates[dump_id] = wikidata_qid
            direct_matches += 1
            continue

        names = []
        for m in _ENRICH_RE_SCHEMA_NAME.finditer(text):
            if m.group(1) == dump_id:
                names.append(m.group(2))
        for m in _ENRICH_RE_SCHEMA_ALT_NAME.finditer(text):
            if m.group(1) == dump_id:
                names.append(m.group(2))
        matched_ids = set()
        for name in names:
            matched_ids.update(name_to_ids.get(name, set()))
        if matched_ids:
            name_matches += 1
            for vid in matched_ids:
                if vid not in wikidata_updates:
                    wikidata_updates[vid] = wikidata_qid

    print(
        f"  [enrich wikidata] Direct: {direct_matches:,}, "
        f"name: {name_matches:,}, unique targets: {len(wikidata_updates):,}"
    )
    updates = 0
    for vid, qid in wikidata_updates.items():
        updates += conn.execute(
            "UPDATE vocabulary SET wikidata_id = ? WHERE id = ? AND wikidata_id IS NULL",
            (qid, vid),
        ).rowcount
    conn.commit()
    print(f"  [enrich wikidata] Updated {updates:,} wikidata_id values")


_ENRICH_CLASSIFICATION_FROM_PRED = {
    "name": "schema_name",
    "alternateName": "schema_alt_name",
}


def _iter_dump_alt_names(dump_dir: Path):
    """Yield (entity_id, name, classification) tuples from a per-entity NT dump dir.

    classification ∈ {'schema_name', 'schema_alt_name'}. Works for any
    Schema.org-shaped dump (person, organisation, topical_term).
    """
    files = os.listdir(dump_dir)
    for i, fname in enumerate(files):
        if i % 20000 == 0 and i > 0:
            print(f"    ... scanned {i:,}/{len(files):,}", flush=True)
        fpath = dump_dir / fname
        try:
            text = fpath.read_text()
        except Exception:
            continue
        entity_id = fname
        for m in _ENRICH_RE_SCHEMA_NAME_OR_ALT.finditer(text):
            if m.group(1) == entity_id:
                yield (entity_id, m.group(3),
                       _ENRICH_CLASSIFICATION_FROM_PRED[m.group(2)])


def _executemany_chunked(conn, sql, rows, chunk_size=500):
    """Stream a row iterable through executemany in fixed-size chunks.

    Returns the count of rows actually inserted (rowcount sum across
    chunks). For INSERT OR IGNORE, this is the number of new rows after
    UNIQUE-constraint dedup.
    """
    inserted = 0
    buf: list = []
    for row in rows:
        buf.append(row)
        if len(buf) >= chunk_size:
            cur = conn.executemany(sql, buf)
            inserted += cur.rowcount if cur.rowcount > 0 else 0
            buf.clear()
    if buf:
        cur = conn.executemany(sql, buf)
        inserted += cur.rowcount if cur.rowcount > 0 else 0
    return inserted


def enrich_person_names_from_dump(conn: sqlite3.Connection, dumps_dir: Path) -> None:
    """Dump-side person_names enrichment for the 31xxx Schema.org cohort.

    Phase 2's LA-shape `identified_by[]` extractor only succeeds for the
    21xxx (LA-shape) cohort. The 31xxx (Schema.org-shape) persons need
    the schema:name + schema:alternateName triples from the person dump
    instead — runs in seconds vs. 8.3h for the equivalent API re-run.

    Idempotent — UNIQUE(person_id, name, lang) on person_names dedupes
    against any rows already inserted by the LA path or prior enrichment.
    """
    dump_dir = dumps_dir / ENRICH_PERSON_DIR
    if not dump_dir.exists():
        extracted = extract_dump("person")
        if extracted is None:
            print("  [enrich person_names] person dump not present — skipping.")
            return
        dump_dir = extracted

    try:
        conn.execute("SELECT 1 FROM person_names LIMIT 1").fetchone()
    except sqlite3.OperationalError:
        print("  [enrich person_names] person_names table not present — skipping.")
        return

    vocab_person_ids = {
        r[0] for r in conn.execute(
            "SELECT id FROM vocabulary WHERE type='person'"
        ).fetchall()
    }
    print(f"  [enrich person_names] Vocab persons: {len(vocab_person_ids):,}")

    skipped_not_in_vocab = 0

    def _row_iter():
        nonlocal skipped_not_in_vocab
        for entity_id, name, classification in _iter_dump_alt_names(dump_dir):
            if entity_id not in vocab_person_ids:
                skipped_not_in_vocab += 1
                continue
            yield (entity_id, name, classification)

    inserted = _executemany_chunked(
        conn,
        "INSERT OR IGNORE INTO person_names (person_id, name, lang, classification) "
        "VALUES (?, ?, NULL, ?)",
        _row_iter(),
    )
    conn.commit()
    print(
        f"  [enrich person_names] Inserted {inserted:,} new rows "
        f"(skipped {skipped_not_in_vocab:,} not in vocabulary)"
    )


def enrich_entity_alt_names_from_dump(
    conn: sqlite3.Connection,
    dumps_dir: Path,
    entity_type: str,
    dump_name: str | None = None,
) -> None:
    """Populate entity_alt_names from a Schema.org-shaped dump.

    ``entity_type`` is the value stored in entity_alt_names.entity_type so
    downstream queries can scope by type. ``dump_name`` defaults to
    ``entity_type`` (the dump filename and entity-type label happen to
    coincide for organisation, topical_term, person, place); pass it
    explicitly only when the dump file's bare name diverges.
    """
    if dump_name is None:
        dump_name = entity_type
    dump_dir = dumps_dir / f"{dump_name}_extracted"
    if not dump_dir.exists():
        extracted = extract_dump(dump_name)
        if extracted is None:
            print(f"  [enrich entity_alt_names:{entity_type}] {dump_name} dump not present — skipping.")
            return
        dump_dir = extracted

    vocab_ids = {
        r[0] for r in conn.execute(
            "SELECT id FROM vocabulary WHERE type=?", (entity_type,)
        ).fetchall()
    }
    print(
        f"  [enrich entity_alt_names:{entity_type}] Vocab entities: {len(vocab_ids):,}"
    )

    skipped_not_in_vocab = 0

    def _row_iter():
        nonlocal skipped_not_in_vocab
        for entity_id, name, classification in _iter_dump_alt_names(dump_dir):
            if entity_id not in vocab_ids:
                skipped_not_in_vocab += 1
                continue
            yield (entity_id, entity_type, name, classification)

    inserted = _executemany_chunked(
        conn,
        "INSERT OR IGNORE INTO entity_alt_names "
        "(entity_id, entity_type, name, lang, classification) "
        "VALUES (?, ?, ?, NULL, ?)",
        _row_iter(),
    )
    conn.commit()
    print(
        f"  [enrich entity_alt_names:{entity_type}] Inserted {inserted:,} rows "
        f"(skipped {skipped_not_in_vocab:,} not in vocabulary)"
    )


def reclassify_external_id_authorities(conn: sqlite3.Connection) -> None:
    """Backfill rijks_internal + handle authority buckets on existing rows.

    The new needles in `_AUTHORITY_NEEDLES` cover fresh inserts; this
    one-shot UPDATE reclassifies any rows previously bucketed as 'other'
    that now match the extended pattern set. Reduces the size of the
    opaque 'other' bucket.
    """
    cur = conn.cursor()
    moved_internal = cur.execute(
        "UPDATE vocabulary_external_ids "
        "SET authority = 'rijks_internal' "
        "WHERE authority = 'other' AND uri LIKE 'https://id.rijksmuseum.nl/%'"
    ).rowcount
    moved_handle = cur.execute(
        "UPDATE vocabulary_external_ids "
        "SET authority = 'handle' "
        "WHERE authority = 'other' AND uri LIKE 'https://hdl.handle.net/%'"
    ).rowcount
    conn.commit()
    if moved_internal or moved_handle:
        print(
            f"  [reclassify authorities] rijks_internal: {moved_internal:,}, "
            f"handle: {moved_handle:,}"
        )


def enrich_concepts_from_skos(conn: sqlite3.Connection, dumps_dir: Path) -> None:
    """Enrich classification concepts with broader_id from 2019 SKOS thesaurus.

    Matches via AAT exactMatch first, then label_nl fallback (unambiguous only).
    """
    zip_path = dumps_dir / ENRICH_SKOS_THESAURUS_ZIP
    if not zip_path.exists():
        print(f"  [enrich concepts] {zip_path} not found — skipping.")
        return

    aat_to_vocab = {}
    for vid, ext_id in conn.execute(
        "SELECT id, external_id FROM vocabulary "
        "WHERE external_id LIKE 'http://vocab.getty.edu/aat/%'"
    ):
        aat_to_vocab[ext_id] = vid

    label_to_vocab = defaultdict(set)
    for vid, label in conn.execute(
        "SELECT id, label_nl FROM vocabulary "
        "WHERE type='classification' AND label_nl IS NOT NULL"
    ):
        label_to_vocab[label].add(vid)
    print(
        f"  [enrich concepts] AAT lookup: {len(aat_to_vocab):,}, "
        f"label lookup: {len(label_to_vocab):,}"
    )

    concepts = []
    with zipfile.ZipFile(zip_path) as z:
        with z.open(z.namelist()[0]) as f:
            context = ET.iterparse(f, events=("end",))
            for _event, elem in context:
                if elem.tag == f"{{{_ENRICH_NS['skos']}}}Concept":
                    about = elem.get(f"{{{_ENRICH_NS['rdf']}}}about", "")
                    thesaurus_id = (
                        about.rsplit(".", 1)[-1] if "THESAU." in about else None
                    )
                    pref_nl = None
                    broader_handle = None
                    exact_match_aat = None
                    for child in elem:
                        if child.tag == f"{{{_ENRICH_NS['skos']}}}prefLabel":
                            lang = child.get(
                                "{http://www.w3.org/XML/1998/namespace}lang", ""
                            )
                            if lang == "nl" and child.text:
                                pref_nl = child.text
                        elif child.tag == f"{{{_ENRICH_NS['skos']}}}broader":
                            broader_handle = child.get(
                                f"{{{_ENRICH_NS['rdf']}}}resource", ""
                            )
                        elif child.tag == f"{{{_ENRICH_NS['skos']}}}exactMatch":
                            uri = child.get(
                                f"{{{_ENRICH_NS['rdf']}}}resource", ""
                            )
                            if "vocab.getty.edu/aat/" in uri:
                                exact_match_aat = uri
                    if broader_handle:
                        broader_thesaurus_id = (
                            broader_handle.rsplit(".", 1)[-1]
                            if "THESAU." in broader_handle
                            else None
                        )
                        concepts.append(
                            (thesaurus_id, pref_nl, exact_match_aat, broader_thesaurus_id)
                        )
                    elem.clear()
    print(f"  [enrich concepts] Parsed {len(concepts):,} concepts with broader links")

    thesaurus_to_vocab = {}
    for thesaurus_id, pref_nl, exact_match_aat, _ in concepts:
        if not thesaurus_id:
            continue
        if exact_match_aat and exact_match_aat in aat_to_vocab:
            thesaurus_to_vocab[thesaurus_id] = aat_to_vocab[exact_match_aat]
        elif pref_nl and pref_nl in label_to_vocab:
            ids = label_to_vocab[pref_nl]
            if len(ids) == 1:
                thesaurus_to_vocab[thesaurus_id] = next(iter(ids))
    print(
        f"  [enrich concepts] Thesaurus→vocab mapping: {len(thesaurus_to_vocab):,}"
    )

    updates = 0
    for thesaurus_id, _, _, broader_thesaurus_id in concepts:
        if not thesaurus_id or not broader_thesaurus_id:
            continue
        vocab_id = thesaurus_to_vocab.get(thesaurus_id)
        broader_vocab_id = thesaurus_to_vocab.get(broader_thesaurus_id)
        if vocab_id and broader_vocab_id:
            updates += conn.execute(
                "UPDATE vocabulary SET broader_id = ? WHERE id = ? AND broader_id IS NULL",
                (broader_vocab_id, vocab_id),
            ).rowcount
    conn.commit()
    print(f"  [enrich concepts] Updated {updates:,} broader_id values")


def _compute_areal_parents(conn: sqlite3.Connection, threshold_km: float = 75.0
                           ) -> list[str]:
    """Identify parents whose geocoded children span ≥ threshold_km pairwise.

    Replaces the SQL GROUP BY HAVING approach which used naive
    ``MAX(lon) - MIN(lon)`` and returned 360° for antimeridian-straddling
    sets (Fiji: spurious 40,051 km diagonal per 2026-04-17 findings, #255).
    """
    from lib.geo_math import max_pairwise_km

    rows = conn.execute(
        """SELECT p.id AS parent_id, c.lat AS c_lat, c.lon AS c_lon
           FROM vocabulary p
           JOIN vocabulary c
             ON c.broader_id = p.id
            AND c.type = 'place'
            AND c.lat IS NOT NULL
           WHERE p.type = 'place' AND p.lat IS NOT NULL"""
    ).fetchall()

    children: dict[str, list[tuple[float, float]]] = defaultdict(list)
    for r in rows:
        children[r["parent_id"]].append((r["c_lat"], r["c_lon"]))

    return [pid for pid, pts in children.items()
            if len(pts) >= 2
            and max_pairwise_km(pts, early_exit_km=threshold_km) >= threshold_km]


_LAYER_B_CITY_PT = frozenset({
    "http://vocab.getty.edu/aat/300008389",  # cities
    "http://vocab.getty.edu/aat/300008375",  # towns
    "http://vocab.getty.edu/aat/300008372",  # villages
    "http://vocab.getty.edu/aat/300008393",  # hamlets
    "http://www.wikidata.org/entity/Q515",      # city
    "http://www.wikidata.org/entity/Q532",      # village
    "http://www.wikidata.org/entity/Q3957",     # town
    "http://www.wikidata.org/entity/Q5084",     # hamlet
    "http://www.wikidata.org/entity/Q1549591",  # metropolis
    "http://www.wikidata.org/entity/Q484170",   # commune of France
    "http://www.wikidata.org/entity/Q747074",   # comune of Italy
    "http://www.wikidata.org/entity/Q486972",   # human settlement (umbrella)
})
# Reserved for future expansion of the allow-list to country-class URIs.
# Layer B's locked allow-list (decisions doc §502) is settlement-only, so
# this set is empty today — but the constant em.COUNTRY_FALLBACK is wired
# into the SQL CASE so flipping a country URI on means just adding it here.
_LAYER_B_COUNTRY_PT: frozenset[str] = frozenset()


def propagate_place_coordinates(conn: sqlite3.Connection) -> None:
    """Propagate lat/lon from geocoded parent places to ungeocoded children.

    Walks vocabulary.broader_id up to depth 10. Idempotent.

    #218 + #262 Layer B (v0.25) + #268: inherited coords are tagged
    ``coord_method = 'inferred'`` with one of three details depending on
    parent placetype (city / country / generic). Layer B's fail-closed
    allow-list (decisions doc §502, mirrored in
    ``em.INHERITANCE_ALLOWED_PLACETYPES``) refuses inheritance unless the
    parent's placetype is on the locked URI set — eliminating the
    (53.0, -2.0) UK-cluster false-positive class as a side-effect.

    Areal-parent filter (post-v0.24, see offline issues #254/#255): a
    parent is treated as areal (excluded from inheritance) when its
    ``is_areal=1`` is set OR when its geocoded children's pairwise
    great-circle distance exceeds 75 km. Pairwise haversine vs. SQL
    bbox-diagonal is antimeridian-safe (Fiji/Oceania fix, #255).

    Blocker sources unioned into ``_areal_parents``:
    1. ``vocabulary.is_areal = 1`` — principled signal from #254.
    2. Spread heuristic — pairwise haversine ≥ 75 km.

    Allow-list source (Layer B) seeded into ``_allowed_parents``:
    parents whose placetype URI is in
    ``em.INHERITANCE_ALLOWED_PLACETYPES``. NULL/missing placetypes fail
    closed (refused).
    """
    conn.execute("DROP TABLE IF EXISTS _areal_parents")
    conn.execute("CREATE TEMP TABLE _areal_parents (id TEXT PRIMARY KEY)")

    # Source 1: rows flagged is_areal=1 by placetype harvest (#254)
    r = conn.execute("""
        INSERT OR IGNORE INTO _areal_parents (id)
        SELECT id FROM vocabulary WHERE type='place' AND is_areal = 1
    """)
    n_placetype = r.rowcount

    # Source 2: spread-based heuristic (#218 original logic)
    areal_ids = _compute_areal_parents(conn, threshold_km=75.0)
    n_spread = 0
    if areal_ids:
        cur = conn.executemany(
            "INSERT OR IGNORE INTO _areal_parents (id) VALUES (?)",
            [(aid,) for aid in areal_ids],
        )
        n_spread = cur.rowcount
    conn.execute(
        "CREATE UNIQUE INDEX IF NOT EXISTS idx_areal_parents_id ON _areal_parents(id)"
    )
    n_areal = conn.execute("SELECT COUNT(*) FROM _areal_parents").fetchone()[0]
    print(f"  [enrich coords] {n_areal:,} parents flagged as areal "
          f"(placetype: {n_placetype:,}, spread ≥ 75 km: +{n_spread:,} new) — "
          f"blocked from inheritance")

    # Layer B: seed the fail-closed allow-list
    conn.execute("DROP TABLE IF EXISTS _allowed_parents")
    conn.execute("CREATE TEMP TABLE _allowed_parents (id TEXT PRIMARY KEY)")
    placeholders = ",".join("?" * len(em.INHERITANCE_ALLOWED_PLACETYPES))
    conn.execute(
        f"INSERT OR IGNORE INTO _allowed_parents (id) "
        f"SELECT id FROM vocabulary "
        f" WHERE type='place' AND placetype IN ({placeholders})",
        tuple(em.INHERITANCE_ALLOWED_PLACETYPES),
    )
    n_allowed = conn.execute("SELECT COUNT(*) FROM _allowed_parents").fetchone()[0]
    print(f"  [enrich coords] Layer B: {n_allowed:,} parents on the "
          f"fail-closed allow-list (settlement-class placetypes)")

    city_pt_list = ",".join("?" * len(_LAYER_B_CITY_PT))
    country_pt_list = ",".join("?" * len(_LAYER_B_COUNTRY_PT)) if _LAYER_B_COUNTRY_PT else "NULL"
    case_expr = f"""
        CASE WHEN p.placetype IN ({city_pt_list}) THEN ?
             WHEN p.placetype IN ({country_pt_list}) THEN ?
             ELSE ?
        END
    """
    case_params = (
        *_LAYER_B_CITY_PT, em.CITY_FALLBACK,
        *_LAYER_B_COUNTRY_PT, em.COUNTRY_FALLBACK,
        em.PARENT_FALLBACK,
    )

    total = 0
    max_depth = 10
    for depth in range(1, max_depth + 1):
        cur = conn.execute(
            f"""UPDATE vocabulary SET
                  lat = (SELECT p.lat FROM vocabulary p WHERE p.id = vocabulary.broader_id),
                  lon = (SELECT p.lon FROM vocabulary p WHERE p.id = vocabulary.broader_id),
                  coord_method = ?,
                  coord_method_detail = (
                      SELECT {case_expr}
                        FROM vocabulary p WHERE p.id = vocabulary.broader_id
                  )
                WHERE type = 'place'
                  AND lat IS NULL
                  AND broader_id IS NOT NULL
                  AND broader_id NOT IN (SELECT id FROM _areal_parents)
                  AND broader_id IN (SELECT id FROM _allowed_parents)
                  AND EXISTS (
                      SELECT 1 FROM vocabulary p
                      WHERE p.id = vocabulary.broader_id AND p.lat IS NOT NULL
                  )
            """,
            (em.tier_for(em.PARENT_FALLBACK), *case_params),
        )
        inherited = cur.rowcount
        if inherited == 0:
            break
        print(f"  [enrich coords] Depth {depth}: {inherited:,} places inherited")
        total += inherited
    conn.commit()
    conn.execute("DROP TABLE IF EXISTS _areal_parents")
    conn.execute("DROP TABLE IF EXISTS _allowed_parents")
    print(f"  [enrich coords] Total inherited: {total:,}")


def log_enrichment_stats(conn: sqlite3.Connection) -> None:
    """Print coverage stats for enriched columns (informational — not an audit)."""
    for col in ("birth_year", "death_year", "gender", "wikidata_id"):
        count = conn.execute(
            f"SELECT COUNT(*) FROM vocabulary WHERE {col} IS NOT NULL"
        ).fetchone()[0]
        print(f"  [enrich stats] {col}: {count:,} non-NULL")
    places_with_broader = conn.execute(
        "SELECT COUNT(*) FROM vocabulary WHERE type='place' AND broader_id IS NOT NULL"
    ).fetchone()[0]
    places_with_coords = conn.execute(
        "SELECT COUNT(*) FROM vocabulary WHERE type='place' AND lat IS NOT NULL"
    ).fetchone()[0]
    total_places = conn.execute(
        "SELECT COUNT(*) FROM vocabulary WHERE type='place'"
    ).fetchone()[0]
    print(
        f"  [enrich stats] Places: {places_with_broader:,}/{total_places:,} "
        f"broader_id, {places_with_coords:,}/{total_places:,} lat"
    )
    cycles = conn.execute(
        "SELECT COUNT(*) FROM vocabulary WHERE broader_id = id"
    ).fetchone()[0]
    if cycles:
        print(f"  [enrich stats] WARNING: {cycles} self-referencing broader_id")


def _create_enrichment_indexes(conn: sqlite3.Connection) -> None:
    """Create partial indexes on enriched columns, gated on data presence.

    `idx_vocab_broader_id` is created earlier in run_phase3 (Part 1 of #242)
    so it is not re-created here.
    """
    index_specs = [
        (
            "SELECT COUNT(*) FROM vocabulary WHERE gender IS NOT NULL",
            "idx_vocab_gender",
            "CREATE INDEX IF NOT EXISTS idx_vocab_gender ON vocabulary(gender) "
            "WHERE gender IS NOT NULL",
        ),
        (
            "SELECT COUNT(*) FROM vocabulary WHERE birth_year IS NOT NULL",
            "idx_vocab_birth_year",
            "CREATE INDEX IF NOT EXISTS idx_vocab_birth_year ON vocabulary(birth_year) "
            "WHERE birth_year IS NOT NULL",
        ),
        (
            "SELECT COUNT(*) FROM vocabulary WHERE wikidata_id IS NOT NULL",
            "idx_vocab_wikidata",
            "CREATE INDEX IF NOT EXISTS idx_vocab_wikidata ON vocabulary(wikidata_id) "
            "WHERE wikidata_id IS NOT NULL",
        ),
    ]
    for count_sql, name, ddl in index_specs:
        count = conn.execute(count_sql).fetchone()[0]
        if count > 0:
            conn.execute(ddl)
            print(f"  [enrich indexes] Created {name} ({count:,} qualifying rows)")
        else:
            print(f"  [enrich indexes] Skipping {name} (no data)")
    conn.commit()


def run_enrichment(conn: sqlite3.Connection, dumps_dir: Path) -> bool:
    """Run the full vocab enrichment pipeline from offline dumps.

    Returns True if any enrichment phase actually executed against a dump,
    False if dumps_dir is missing entirely (so the caller can skip audit
    thresholds that only make sense after enrichment ran).

    Idempotent: every UPDATE is gated on `<col> IS NULL`, so re-running
    against an already-enriched DB is a no-op.
    """
    if not dumps_dir.exists():
        print(
            f"  [enrichment] Dumps dir not found: {dumps_dir}\n"
            f"  [enrichment] Skipping — DB will still open (column stubs present)\n"
            f"  [enrichment] but creator wikidata and place hierarchy will be empty."
        )
        return False

    print(f"  [enrichment] Using dumps dir: {dumps_dir}")
    t0 = time.time()
    enrich_places_from_dump(conn, dumps_dir)
    enrich_actors_from_edm(conn, dumps_dir)
    enrich_wikidata_from_person_dump(conn, dumps_dir)
    # Dump-side person_names enrichment (covers the 31xxx Schema.org cohort).
    # Runs AFTER enrich_wikidata_from_person_dump so the latter still uses the
    # pre-existing person_names corpus for name-based wikidata matching.
    enrich_person_names_from_dump(conn, dumps_dir)
    enrich_entity_alt_names_from_dump(conn, dumps_dir, "organisation")
    enrich_concepts_from_skos(conn, dumps_dir)
    # Reclassify rijks_internal + handle out of the 'other' bucket. Runs after
    # all enrich_* steps so newly-inserted rows are also reclassified.
    reclassify_external_id_authorities(conn)
    propagate_place_coordinates(conn)
    _create_enrichment_indexes(conn)

    # Ensure version_info exists — enrichment runs before the main version_info
    # block later in run_phase3, so we can't assume the table is already there.
    conn.execute(
        "CREATE TABLE IF NOT EXISTS version_info (key TEXT PRIMARY KEY, value TEXT)"
    )
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    conn.execute(
        "INSERT OR REPLACE INTO version_info (key, value) VALUES ('enriched_at', ?)",
        (now,),
    )
    conn.execute(
        "INSERT OR REPLACE INTO version_info (key, value) "
        "VALUES ('enrichment_sources', ?)",
        ("2019-edm-actors+2019-skos-thesaurus+2025-person-dump+2025-place-dump",),
    )
    conn.commit()

    log_enrichment_stats(conn)
    print(f"  [enrichment] Complete in {time.time() - t0:.1f}s")
    return True


# ─── Phase 3: Validation ─────────────────────────────────────────────

def run_phase3(
    conn: sqlite3.Connection,
    geo_csv: str | None = None,
    audit_results: dict[str, list[AuditResult]] | None = None,
    skip_enrichment: bool = False,
    dumps_dir: Path | None = None,
):
    """Phase 3: Post-processing (geocoding import, enrichment, FTS, stats).

    ``audit_results`` is the run-wide accumulator threaded through from main();
    when present, two audit checkpoints fire — once after import_geocoding
    (catches #218 ``geocode_method`` regressions) and once at end of phase 3
    (catches downstream zero-row regressions like #220 → artwork_exhibitions).

    Enrichment (#242 part 3) folds the former ``enrich-vocab-from-dumps.py``
    pipeline into harvest. It runs after mappings/rights normalization so
    ``vocabulary.id`` is stable, and before the FTS / conditional-index /
    VACUUM blocks so the inherited place coordinates are captured by the
    ``idx_vocab_lat_lon`` partial index. Pass ``skip_enrichment=True`` to
    bypass (the DB still opens because the column stubs are added below).
    """
    cur = conn.cursor()

    # geocode_method column must exist before import_geocoding() writes to it
    # and before the phase3.geocoding audit fires.
    # NOTE: geocode_method is deprecated as of #218 in favor of the three
    # per-field method columns below (coord_method, external_id_method,
    # broader_method). Kept for schema compatibility during migration; to be
    # dropped in a later cleanup pass.
    vocab_cols = get_columns(conn, "vocabulary")
    if "geocode_method" not in vocab_cols:
        conn.execute("ALTER TABLE vocabulary ADD COLUMN geocode_method TEXT")
        conn.commit()

    # #218 / #268: per-field enrichment provenance. NULL = as-is from Rijksmuseum
    # (LDES may overwrite); 'deterministic' = our direct canonical lookup
    # (Wikidata P625, GeoNames, Getty TGN) — preserve across LDES updates;
    # 'inferred' = our heuristic (reconciliation, parent-fallback, validation
    # fix) — preserve; 'manual' = curator review — never auto-overwrite.
    # The coarse tier names are shared across geo, provenance, and alt-label
    # audit trails — see scripts/enrichment_tiers.py. The fine detail values
    # live in scripts/enrichment_methods.py.
    enrichment_method_cols = ("coord_method", "external_id_method", "broader_method")
    for col_name in enrichment_method_cols:
        if col_name not in vocab_cols:
            conn.execute(f"ALTER TABLE vocabulary ADD COLUMN {col_name} TEXT")
    conn.commit()

    # #218 fine-grained detail — paired with the coarse method columns above.
    # coord_method stores the coarse tier ('deterministic' / 'inferred' / 'manual'
    # — see #268; legacy 'authority' / 'derived' / 'human' fully migrated),
    # *_detail stores the specific phase identifier ('wikidata_p625',
    # 'geonames_api', 'whg_reconciliation', 'parent_fallback', etc. from
    # scripts/enrichment_methods.py). Both layers live in the DB so the
    # v0.24 backfill CSV export can surface them for downstream consumers.
    enrichment_detail_cols = (
        "coord_method_detail", "external_id_method_detail", "broader_method_detail",
    )
    for col_name in enrichment_detail_cols:
        if col_name not in vocab_cols:
            conn.execute(f"ALTER TABLE vocabulary ADD COLUMN {col_name} TEXT")
    conn.commit()

    # #254: placetype provenance for the "centroid not meaningful for
    # point-based queries" signal. Covers both (a) admin polygons with
    # wide trimmed spread and (b) region-scale entities (oceans, empires,
    # historical colonial territories) where no centroid is useful.
    #   placetype        — AAT URI (TGN) or Wikidata QID
    #   placetype_source — 'tgn' | 'wikidata' | 'manual' | 'label_heuristic'
    #   is_areal         — 1 | 0 | NULL (NULL = we have no opinion yet)
    # Populated by scripts/harvest-placetypes.py (SPARQL side-pass) and
    # scripts/geocoding/apply_areal_overrides.py (manual TSV). Side-pass is idempotent
    # via WHERE placetype_source IS NULL guards; manual overrides take
    # precedence over authority-sourced values. Survives cold-reset of
    # lat/lon/coord_method since these are attributes of the entity, not
    # of any given geocoding run.
    placetype_cols = (
        ("placetype",        "TEXT"),
        ("placetype_source", "TEXT"),
        ("is_areal",         "INTEGER"),
    )
    for col_name, col_type in placetype_cols:
        if col_name not in vocab_cols:
            conn.execute(f"ALTER TABLE vocabulary ADD COLUMN {col_name} {col_type}")
    conn.commit()

    # Person-enrichment columns must exist at server open() time — VocabularyDb.ts
    # caches `stmtArtworkMappings` against v.birth_year/death_year/gender/wikidata_id
    # unconditionally. Populated later by enrich-vocab-from-dumps.py (Phases 2a/2b);
    # leaving them NULL is sufficient for the server to open the DB. See issue #242.
    person_enrichment_cols = [
        ("birth_year", "INTEGER"),
        ("death_year", "INTEGER"),
        ("gender", "TEXT"),
        ("wikidata_id", "TEXT"),
    ]
    for col_name, col_type in person_enrichment_cols:
        if col_name not in vocab_cols:
            conn.execute(f"ALTER TABLE vocabulary ADD COLUMN {col_name} {col_type}")
    conn.commit()

    # broader_id index: silences the `expandPlaceHierarchy will be slow` startup
    # warning and makes nearPlace / place-hierarchy expansion fast on freshly-
    # harvested DBs. Partial index is safe even when broader_id is all NULL
    # (populated by enrich-vocab-from-dumps.py Phases 2c/2d). See issue #242.
    if "broader_id" in vocab_cols:
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_vocab_broader_id "
            "ON vocabulary(broader_id) WHERE broader_id IS NOT NULL"
        )
        conn.commit()

    # Import geocoding data if CSV provided
    if geo_csv:
        print("\n--- Geocoding Import ---")
        import_geocoding(conn, geo_csv)
        print()
        if audit_results is not None:
            geo_audit = run_phase_audit(conn, "phase3.geocoding")
            audit_results["phase3.geocoding"] = geo_audit
            format_stdout_table(geo_audit, "phase3.geocoding")

    # ── New Phase 3 tables (art_id-independent) ──

    # Museum rooms reference table (static data from #212 crawl)
    print("\n--- Museum Rooms ---")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS museum_rooms (
            room_hash  TEXT PRIMARY KEY,
            room_id    TEXT NOT NULL,
            floor      TEXT,
            room_name  TEXT
        )
    """)
    # #229 Part A: seed file now lives under data/seed/ so it can be
    # committed to the repo without conflicting with the gitignored data/
    # convention. Fall back to the legacy location for backwards compat
    # with any operator still pointing at the old path.
    rooms_json = PROJECT_DIR / "data" / "seed" / "museum-rooms.json"
    if not rooms_json.exists():
        legacy = PROJECT_DIR / "data" / "museum-rooms.json"
        if legacy.exists():
            rooms_json = legacy
    if rooms_json.exists():
        rooms_data = json.loads(rooms_json.read_text(encoding="utf-8"))
        for room in rooms_data:
            conn.execute(
                "INSERT OR IGNORE INTO museum_rooms (room_hash, room_id, floor, room_name) VALUES (?, ?, ?, ?)",
                (room.get("room_hash", ""), room["room_id"], room.get("floor"), room.get("room_name")),
            )
        conn.commit()
        print(f"  Seeded {len(rooms_data)} museum rooms from {rooms_json}")
    else:
        print(f"  Skipping room seeding ({rooms_json} not found)")

    # sync_state table for LDES consumer (#205)
    print("\n--- Sync State ---")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS sync_state (
            key   TEXT PRIMARY KEY,
            value TEXT
        )
    """)
    conn.commit()
    print("  sync_state table ready")

    # Convert dimension 0.0 sentinels → NULL (#195)
    print("\n--- Dimension Sentinel Cleanup ---")
    for col in ["height_cm", "width_cm", "depth_cm", "diameter_cm"]:
        if col in get_columns(conn, "artworks"):
            updated = conn.execute(f"UPDATE artworks SET {col} = NULL WHERE {col} = 0.0").rowcount
            if updated > 0:
                print(f"  {col}: {updated:,} zero sentinels → NULL")
    conn.commit()

    # ── Integer-encode mappings & normalize rights for ~1.2 GB savings ──
    print("\n--- Mappings Normalization ---")
    normalize_mappings(conn)
    normalize_rights(conn)

    # ── Vocab enrichment from offline dumps (#242 part 3) ──
    print("\n--- Vocabulary Enrichment ---")
    if skip_enrichment:
        print("  Skipping enrichment (--skip-enrichment). DB will still open")
        print("  (column stubs present), but creator wikidata and place")
        print("  hierarchy will be empty.")
    else:
        run_enrichment(conn, dumps_dir or DUMPS_DIR)

    # #268: tag broader_method on every row that has a broader_id but no tier
    # tag yet. All three broader_id write sites (baseline LA harvest, place dump
    # enrichment, SKOS thesaurus enrichment) are authoritative — they pull from
    # external curator-stamped sources — so 'deterministic' is correct. Runs
    # unconditionally (covers the --skip-enrichment baseline-only path too).
    tagged = conn.execute(
        "UPDATE vocabulary SET broader_method = 'deterministic' "
        "WHERE broader_id IS NOT NULL AND broader_method IS NULL"
    ).rowcount
    conn.commit()
    print(f"  [enrich tier] Tagged {tagged:,} rows broader_method='deterministic'")

    # ── Post-normalization joins (require art_id from normalize_mappings) ──

    # artwork_exhibitions junction table (from Phase 0 exhibition_members)
    print("\n--- Artwork Exhibitions ---")
    if table_exists(conn, "exhibition_members"):
        conn.execute("""
            CREATE TABLE IF NOT EXISTS artwork_exhibitions (
                art_id        INTEGER NOT NULL,
                exhibition_id INTEGER NOT NULL,
                PRIMARY KEY (art_id, exhibition_id)
            ) WITHOUT ROWID
        """)
        # hmo_id is the numeric suffix of the Linked Art URI (e.g. "200421947"),
        # not the object_number (e.g. "AK-BR-324"). Join via linked_art_uri while
        # it is still present; it is dropped later in Phase 3.
        if "linked_art_uri" in get_columns(conn, "artworks"):
            # Materialise the suffix once so both the INSERT and the CSV export
            # share one full-scan pass instead of two.
            conn.execute("""
                CREATE TEMP TABLE _tmp_hmo_art AS
                SELECT art_id, object_number,
                       SUBSTR(linked_art_uri,
                           INSTR(linked_art_uri, '.nl/') + 4) AS hmo_id
                FROM artworks
                WHERE linked_art_uri IS NOT NULL
                  AND linked_art_uri != ''
                  AND art_id IS NOT NULL
            """)
            conn.execute("CREATE INDEX _tmp_hmo_art_idx ON _tmp_hmo_art(hmo_id)")
            inserted = conn.execute("""
                INSERT OR IGNORE INTO artwork_exhibitions (art_id, exhibition_id)
                SELECT t.art_id, em.exhibition_id
                FROM exhibition_members em
                JOIN _tmp_hmo_art t ON t.hmo_id = em.hmo_id
            """).rowcount
            conn.commit()
            print(f"  artwork_exhibitions: {inserted:,} rows")

            total_ae = cur.execute("SELECT COUNT(*) FROM artwork_exhibitions").fetchone()[0]
            total_em = cur.execute("SELECT COUNT(*) FROM exhibition_members").fetchone()[0]
            if total_ae < total_em:
                print(f"  Note: {total_em - total_ae:,} exhibition memberships could not be resolved to art_ids")

            # Write hmo_id → art_id + object_number mapping CSV while linked_art_uri is
            # still available. Includes unresolved rows (art_id=NULL) so future bugs in
            # artwork_exhibitions can be diagnosed and backfilled without a full re-harvest.
            hmo_map_path = PROJECT_DIR / "data" / "backfills" / "exhibition-hmo-id-map.csv"
            hmo_map_path.parent.mkdir(parents=True, exist_ok=True)
            rows = conn.execute("""
                SELECT em.hmo_id, t.art_id, t.object_number
                FROM exhibition_members em
                LEFT JOIN _tmp_hmo_art t ON t.hmo_id = em.hmo_id
                GROUP BY em.hmo_id
                ORDER BY em.hmo_id
            """).fetchall()
            with open(hmo_map_path, "w", newline="", encoding="utf-8") as f:
                writer = csv.writer(f)
                writer.writerow(["hmo_id", "art_id", "object_number"])
                writer.writerows(rows)
            resolved = sum(1 for r in rows if r[1] is not None)
            print(f"  Wrote {len(rows):,} rows to {hmo_map_path.name} ({resolved:,} resolved, {len(rows)-resolved:,} unresolved)")

            conn.execute("DROP TABLE _tmp_hmo_art")
        else:
            print("  Note: linked_art_uri not available (Phase 3 re-run) — skipping insert")
            inserted = conn.execute("SELECT COUNT(*) FROM artwork_exhibitions").fetchone()[0]
            conn.commit()
            print(f"  artwork_exhibitions: {inserted:,} rows")
    else:
        print("  Skipping (no exhibition_members table)")

    # #251: Resolve via an indexed temp table. The straightforward
    # `WHERE SUBSTR(...) = SUBSTR(...)` is O(N×M) — no index can cover
    # SUBSTR on both join sides. Same pattern as the artwork_exhibitions
    # block above and scripts/finish-v024-phase3.py.
    has_la_uri = "linked_art_uri" in get_columns(conn, "artworks")
    if has_la_uri:
        conn.execute("""
            CREATE TEMP TABLE _tmp_hmo_art AS
            SELECT art_id,
                   SUBSTR(linked_art_uri, INSTR(linked_art_uri, '.nl/') + 4) AS hmo_id
            FROM artworks
            WHERE linked_art_uri IS NOT NULL
              AND linked_art_uri != ''
              AND art_id IS NOT NULL
        """)
        conn.execute("CREATE INDEX _tmp_hmo_art_idx ON _tmp_hmo_art(hmo_id)")

    # Resolve related_objects.related_art_id
    print("\n--- Resolving Related Object Art IDs ---")
    if table_exists(conn, "related_objects"):
        ro_count = cur.execute("SELECT COUNT(*) FROM related_objects").fetchone()[0]
        if ro_count > 0:
            if has_la_uri:
                conn.execute("""
                    UPDATE related_objects SET related_art_id = (
                        SELECT hmo.art_id FROM _tmp_hmo_art hmo
                        WHERE hmo.hmo_id = SUBSTR(related_objects.related_la_uri,
                                                  INSTR(related_objects.related_la_uri, '.nl/') + 4)
                    )
                    WHERE related_art_id IS NULL
                """)
                resolved = cur.execute("SELECT COUNT(*) FROM related_objects WHERE related_art_id IS NOT NULL").fetchone()[0]
                print(f"  Resolved {resolved:,} of {ro_count:,} related objects to art_ids")
            else:
                print(f"  Skipping (linked_art_uri not available — Phase 3 re-run, {ro_count:,} rows untouched)")
        else:
            print("  No related objects to resolve")
        conn.commit()
    else:
        print("  Skipping (no related_objects table)")

    # Resolve artwork_parent.parent_art_id
    print("\n--- Resolving Artwork Parent Art IDs ---")
    if table_exists(conn, "artwork_parent"):
        ap_count = cur.execute("SELECT COUNT(*) FROM artwork_parent").fetchone()[0]
        if ap_count > 0:
            if has_la_uri:
                conn.execute("""
                    UPDATE artwork_parent SET parent_art_id = (
                        SELECT hmo.art_id FROM _tmp_hmo_art hmo
                        WHERE hmo.hmo_id = SUBSTR(artwork_parent.parent_la_uri,
                                                  INSTR(artwork_parent.parent_la_uri, '.nl/') + 4)
                    )
                    WHERE parent_art_id IS NULL
                """)
                resolved = cur.execute("SELECT COUNT(*) FROM artwork_parent WHERE parent_art_id IS NOT NULL").fetchone()[0]
                print(f"  Resolved {resolved:,} of {ap_count:,} parent links to art_ids")
            else:
                print(f"  Skipping (linked_art_uri not available — Phase 3 re-run, {ap_count:,} rows untouched)")
        else:
            print("  No parent links to resolve")
        conn.commit()
    else:
        print("  Skipping (no artwork_parent table)")

    if has_la_uri:
        conn.execute("DROP TABLE _tmp_hmo_art")

    # #253: preserve (art_id → hmo_id) mapping before the linked_art_uri column is
    # dropped, so decoupled post-harvest backfills (e.g. VI-iconclass #203) can still
    # derive per-artwork HMO URIs without a fresh OAI-PMH re-harvest.
    print("\n--- Preserving art_id → hmo_id lookup ---")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS artwork_hmo_ids (
            art_id INTEGER PRIMARY KEY,
            hmo_id TEXT NOT NULL
        )
    """)
    if "linked_art_uri" in get_columns(conn, "artworks"):
        conn.execute("""
            INSERT OR IGNORE INTO artwork_hmo_ids (art_id, hmo_id)
            SELECT art_id,
                   SUBSTR(linked_art_uri, INSTR(linked_art_uri, '.nl/') + 4)
            FROM artworks
            WHERE tier2_done = 1
              AND linked_art_uri IS NOT NULL AND linked_art_uri != ''
              AND art_id IS NOT NULL
        """)
    count = cur.execute("SELECT COUNT(*) FROM artwork_hmo_ids").fetchone()[0]
    print(f"  artwork_hmo_ids: {count:,} rows")
    conn.commit()

    # Drop harvest-only index (only useful during Phase 4 to find unresolved artworks)
    conn.execute("DROP INDEX IF EXISTS idx_artworks_tier2")

    # Drop redundant/harmful indexes (safety — in case running against older DB)
    conn.execute("DROP INDEX IF EXISTS idx_mappings_field_artwork")
    conn.execute("DROP INDEX IF EXISTS idx_mappings_vocab")

    # Drop harvest-only columns (SQLite >= 3.35.0)
    for col in ["linked_art_uri", "tier2_done", "visualitem_uri"]:
        try:
            conn.execute(f"ALTER TABLE artworks DROP COLUMN {col}")
            print(f"    Dropped {col} column")
        except Exception:
            print(f"    Note: Could not drop {col} (SQLite < 3.35.0)")
    conn.commit()

    print("\n" + "=" * 60)
    print("DATABASE STATISTICS")
    print("=" * 60)

    for table in ["vocabulary", "artworks", "mappings"]:
        count = cur.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
        print(f"  {table}: {count:,} rows")

    # New table stats
    for table in ["exhibitions", "artwork_exhibitions", "modifications", "related_objects",
                   "examinations", "title_variants", "assignment_pairs", "artwork_parent"]:
        if table_exists(conn, table):
            count = cur.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
            if count > 0:
                print(f"  {table}: {count:,} rows")

    print("\n--- Vocabulary by Type ---")
    rows = cur.execute("""
        SELECT v.type, COUNT(*) as vocab_cnt,
               COUNT(DISTINCT m.artwork_id) as artwork_cnt
        FROM vocabulary v
        LEFT JOIN mappings m ON m.vocab_rowid = v.vocab_int_id
        GROUP BY v.type
        ORDER BY artwork_cnt DESC
    """).fetchall()
    for vtype, vcnt, acnt in rows:
        print(f"  {vtype:20s} {vcnt:8,} terms    {acnt:8,} artworks linked")

    print("\n--- Mappings by Field ---")
    rows = cur.execute("""
        SELECT f.name as field, COUNT(*) as cnt, COUNT(DISTINCT m.artwork_id) as artworks
        FROM mappings m
        JOIN field_lookup f ON m.field_id = f.id
        GROUP BY m.field_id
        ORDER BY cnt DESC
    """).fetchall()
    for field, cnt, artworks in rows:
        print(f"  {field:20s} {cnt:10,} mappings    {artworks:8,} artworks")

    print("\n--- Vocabulary Coverage ---")
    total_vocab_ids = cur.execute("SELECT COUNT(DISTINCT vocab_rowid) FROM mappings").fetchone()[0]
    # After integer-encoding, all vocab_rowids are valid (orphans dropped during JOIN migration)
    print(f"  Distinct vocab IDs referenced: {total_vocab_ids:,}")
    print(f"  All matched (orphans dropped during integer-encoding)")

    # ── Post-processing: vocab_term_counts, FTS5, collection_set/rights stats ──

    print("\n--- Post-processing ---")

    # Create vocab_term_counts table (preserves text vocab_id for topTermUris())
    print("  Building vocab_term_counts...")
    conn.execute("DROP TABLE IF EXISTS vocab_term_counts")
    conn.execute("""
        CREATE TABLE vocab_term_counts AS
        SELECT v.id AS vocab_id, COUNT(*) AS cnt
        FROM mappings m
        JOIN vocabulary v ON m.vocab_rowid = v.vocab_int_id
        GROUP BY m.vocab_rowid
    """)
    conn.execute("CREATE INDEX idx_vtc_cnt ON vocab_term_counts(cnt DESC)")
    vtc_count = cur.execute("SELECT COUNT(*) FROM vocab_term_counts").fetchone()[0]
    print(f"    vocab_term_counts: {vtc_count:,} rows")
    conn.commit()

    # Create FTS5 virtual table for vocabulary label search
    print("  Building FTS5 index on vocabulary labels...")
    conn.execute("DROP TABLE IF EXISTS vocabulary_fts")
    conn.execute("""
        CREATE VIRTUAL TABLE vocabulary_fts USING fts5(
            label_en, label_nl,
            content='vocabulary', content_rowid='rowid',
            tokenize='unicode61 remove_diacritics 2'
        )
    """)
    conn.execute("INSERT INTO vocabulary_fts(vocabulary_fts) VALUES('rebuild')")
    fts_count = cur.execute("SELECT COUNT(*) FROM vocabulary_fts").fetchone()[0]
    print(f"    vocabulary_fts: {fts_count:,} rows")
    conn.commit()

    # Build FTS5 index on person name variants (if person_names table has data)
    pn_count = cur.execute(
        "SELECT COUNT(*) FROM person_names"
    ).fetchone()[0]
    if pn_count > 0:
        print("  Building FTS5 index on person name variants...")
        conn.execute("DROP TABLE IF EXISTS person_names_fts")
        conn.execute("""
            CREATE VIRTUAL TABLE person_names_fts USING fts5(
                name,
                content='person_names', content_rowid='rowid',
                tokenize='unicode61 remove_diacritics 2'
            )
        """)
        conn.execute("INSERT INTO person_names_fts(person_names_fts) VALUES('rebuild')")
        pnf_count = cur.execute("SELECT COUNT(*) FROM person_names_fts").fetchone()[0]
        print(f"    person_names_fts: {pnf_count:,} rows")
        conn.commit()
    else:
        print("  Skipping person_names_fts (no person name data yet)")

    # Rebuild entity_alt_names_fts (organisations etc.)
    try:
        ean_count = cur.execute("SELECT COUNT(*) FROM entity_alt_names").fetchone()[0]
    except sqlite3.OperationalError:
        ean_count = 0
    if ean_count > 0:
        print("  Rebuilding FTS5 index on entity_alt_names...")
        conn.execute(
            "INSERT INTO entity_alt_names_fts(entity_alt_names_fts) VALUES('rebuild')"
        )
        eanf_count = cur.execute("SELECT COUNT(*) FROM entity_alt_names_fts").fetchone()[0]
        print(f"    entity_alt_names_fts: {eanf_count:,} rows")
        conn.commit()
    else:
        print("  Skipping entity_alt_names_fts (no rows yet)")

    # Populate normalized label columns (space-stripped lowercase for LIKE fallback)
    print("  Populating normalized label columns...")
    conn.execute("""
        UPDATE vocabulary SET
            label_en_norm = REPLACE(LOWER(label_en), ' ', ''),
            label_nl_norm = REPLACE(LOWER(label_nl), ' ', '')
        WHERE label_en IS NOT NULL OR label_nl IS NOT NULL
    """)
    norm_count = cur.execute(
        "SELECT COUNT(*) FROM vocabulary WHERE label_en_norm IS NOT NULL OR label_nl_norm IS NOT NULL"
    ).fetchone()[0]
    print(f"    Normalized labels: {norm_count:,} rows")
    conn.commit()

    # Create FTS5 virtual table for artwork text fields (Tier 2)
    # Detect Tier 2 data by checking for non-null text fields (tier2_done column may be dropped)
    has_tier2 = cur.execute(
        "SELECT COUNT(*) FROM artworks WHERE inscription_text IS NOT NULL OR description_text IS NOT NULL OR narrative_text IS NOT NULL"
    ).fetchone()[0]
    if has_tier2 > 0:
        print("  Building FTS5 index on artwork text fields (Tier 2)...")
        conn.execute("DROP TABLE IF EXISTS artwork_texts_fts")
        conn.execute("""
            CREATE VIRTUAL TABLE artwork_texts_fts USING fts5(
                inscription_text, provenance_text, credit_line, description_text, narrative_text,
                title_all_text,
                content='artworks', content_rowid='rowid',
                tokenize='unicode61 remove_diacritics 2'
            )
        """)
        conn.execute("INSERT INTO artwork_texts_fts(artwork_texts_fts) VALUES('rebuild')")
        atf_count = cur.execute("SELECT COUNT(*) FROM artwork_texts_fts").fetchone()[0]
        print(f"    artwork_texts_fts: {atf_count:,} rows")
        conn.commit()
    else:
        print("  Skipping artwork_texts_fts (no Tier 2 data yet)")

    # Conditional indexes: only create when relevant data exists
    conditional_indexes = [
        (
            "SELECT COUNT(*) FROM artworks WHERE height_cm IS NOT NULL OR width_cm IS NOT NULL",
            "dimension indexes",
            [
                "CREATE INDEX IF NOT EXISTS idx_artworks_height ON artworks(height_cm) WHERE height_cm IS NOT NULL",
                "CREATE INDEX IF NOT EXISTS idx_artworks_width ON artworks(width_cm) WHERE width_cm IS NOT NULL",
            ],
        ),
        (
            "SELECT COUNT(*) FROM artworks WHERE date_earliest IS NOT NULL",
            "date range index",
            [
                "CREATE INDEX IF NOT EXISTS idx_artworks_date_range ON artworks(date_earliest, date_latest) WHERE date_earliest IS NOT NULL",
            ],
        ),
        (
            "SELECT COUNT(*) FROM artworks WHERE record_modified IS NOT NULL",
            "record_modified index",
            [
                "CREATE INDEX IF NOT EXISTS idx_artworks_record_modified ON artworks(record_modified) WHERE record_modified IS NOT NULL",
            ],
        ),
        (
            "SELECT COUNT(*) FROM vocabulary WHERE lat IS NOT NULL",
            "geo index",
            [
                "CREATE INDEX IF NOT EXISTS idx_vocab_lat_lon ON vocabulary(lat, lon) WHERE lat IS NOT NULL",
            ],
        ),
    ]
    for count_sql, label, index_sqls in conditional_indexes:
        count = cur.execute(count_sql).fetchone()[0]
        if count > 0:
            print(f"  Creating {label} — {count:,} qualifying rows...")
            for sql in index_sqls:
                conn.execute(sql)
            conn.commit()
        else:
            print(f"  Skipping {label} (no qualifying data)")

    # Collection set stats
    set_field_id = cur.execute(
        "SELECT id FROM field_lookup WHERE name = 'collection_set'"
    ).fetchone()
    if set_field_id:
        set_mappings = cur.execute(
            "SELECT COUNT(*) FROM mappings WHERE field_id = ?", (set_field_id[0],)
        ).fetchone()[0]
        distinct_sets = cur.execute(
            "SELECT COUNT(DISTINCT vocab_rowid) FROM mappings WHERE field_id = ?", (set_field_id[0],)
        ).fetchone()[0]
        print(f"    collection_set mappings: {set_mappings:,} ({distinct_sets:,} distinct sets)")
    else:
        print(f"    collection_set mappings: 0 (no collection_set field)")

    # Rights coverage (via rights_lookup after normalization)
    rights_total = cur.execute(
        "SELECT COUNT(*) FROM artworks WHERE rights_id IS NOT NULL"
    ).fetchone()[0]
    rights_distinct = cur.execute(
        "SELECT COUNT(DISTINCT rights_id) FROM artworks WHERE rights_id IS NOT NULL"
    ).fetchone()[0]
    print(f"    rights coverage: {rights_total:,} artworks ({rights_distinct:,} distinct URIs)")

    # Show distinct rights URIs
    print("\n--- Rights URIs ---")
    rows = cur.execute("""
        SELECT r.uri, COUNT(*) as cnt
        FROM artworks a
        JOIN rights_lookup r ON a.rights_id = r.id
        GROUP BY a.rights_id
        ORDER BY cnt DESC
    """).fetchall()
    for uri, cnt in rows:
        print(f"  {cnt:8,}  {uri}")

    # Tier 2 stats
    if has_tier2 > 0:
        print(f"\n--- Tier 2 Coverage ({has_tier2:,} artworks with text data) ---")

        # Text and dimension column coverage
        text_cols = [
            ("inscription_text", "Inscriptions"),
            ("provenance_text",  "Provenance"),
            ("credit_line",      "Credit lines"),
            ("description_text", "Descriptions"),
            ("narrative_text",   "Narratives"),
            ("title_all_text",   "All titles"),
        ]
        non_null_cols = [
            ("date_earliest",    "Dates"),
            ("height_cm",        "Height"),
            ("width_cm",         "Width"),
            ("depth_cm",         "Depth"),
            ("diameter_cm",      "Diameter"),
            ("weight_g",         "Weight"),
            ("date_display",     "Date display"),
            ("current_location", "Current loc."),
        ]
        for col, label in text_cols:
            cnt = cur.execute(f"SELECT COUNT(*) FROM artworks WHERE {col} IS NOT NULL AND {col} != ''").fetchone()[0]
            print(f"  {label:20s} {cnt:8,} artworks")
        for col, label in non_null_cols:
            cnt = cur.execute(f"SELECT COUNT(*) FROM artworks WHERE {col} IS NOT NULL").fetchone()[0]
            print(f"  {label:20s} {cnt:8,} artworks")

        # Mapping field coverage
        for field, label in [("production_role", "Prod. roles"), ("attribution_qualifier", "Attr. qualifiers"), ("creator", "Creators")]:
            fid = cur.execute("SELECT id FROM field_lookup WHERE name = ?", (field,)).fetchone()
            if fid:
                cnt = cur.execute(
                    "SELECT COUNT(*) FROM mappings WHERE field_id = ?", (fid[0],)
                ).fetchone()[0]
                artworks = cur.execute(
                    "SELECT COUNT(DISTINCT artwork_id) FROM mappings WHERE field_id = ?", (fid[0],)
                ).fetchone()[0]
            else:
                cnt, artworks = 0, 0
            print(f"  {label:20s} {cnt:8,} mappings ({artworks:,} artworks)")

        # Creator label coverage (should increase from ~55% to ~95%+ after assigned_by extraction)
        creator_label_cnt = cur.execute(
            "SELECT COUNT(*) FROM artworks WHERE creator_label IS NOT NULL AND creator_label != ''"
        ).fetchone()[0]
        total_artworks_cnt = cur.execute("SELECT COUNT(*) FROM artworks").fetchone()[0]
        pct = (creator_label_cnt / total_artworks_cnt * 100) if total_artworks_cnt > 0 else 0
        print(f"  {'Creator labels':20s} {creator_label_cnt:8,} artworks ({pct:.1f}%)")

        # has_image coverage
        artworks_cols = get_columns(conn, "artworks")
        if "has_image" in artworks_cols:
            has_img_cnt = cur.execute("SELECT COUNT(*) FROM artworks WHERE has_image = 1").fetchone()[0]
            pct = (has_img_cnt / total_artworks_cnt * 100) if total_artworks_cnt > 0 else 0
            print(f"  {'Has image':20s} {has_img_cnt:8,} artworks ({pct:.1f}%)")

        # IIIF ID coverage
        if "iiif_id" in artworks_cols:
            iiif_cnt = cur.execute("SELECT COUNT(*) FROM artworks WHERE iiif_id IS NOT NULL").fetchone()[0]
            pct = (iiif_cnt / total_artworks_cnt * 100) if total_artworks_cnt > 0 else 0
            print(f"  {'IIIF ID':20s} {iiif_cnt:8,} artworks ({pct:.1f}%)")

        # Tier 2 pending (only if columns still exist — they're dropped in Phase 3)
        if "tier2_done" in artworks_cols and "linked_art_uri" in artworks_cols:
            tier2_pending = cur.execute(
                "SELECT COUNT(*) FROM artworks WHERE tier2_done = 0 AND linked_art_uri IS NOT NULL AND linked_art_uri != ''"
            ).fetchone()[0]
            if tier2_pending > 0:
                print(f"  Still pending:     {tier2_pending:,} artworks (use --resume --phase 4)")

    print("\n--- Sample Queries ---")

    # Dogs (Iconclass 34B11)
    count = cur.execute("""
        SELECT COUNT(DISTINCT m.artwork_id)
        FROM mappings m
        JOIN vocabulary v ON m.vocab_rowid = v.vocab_int_id
        WHERE v.notation = '34B11'
    """).fetchone()[0]
    print(f"  Iconclass 34B11 (dog): {count:,} artworks")

    # Crucifixion
    count = cur.execute("""
        SELECT COUNT(DISTINCT m.artwork_id)
        FROM mappings m
        JOIN vocabulary v ON m.vocab_rowid = v.vocab_int_id
        WHERE v.label_en LIKE '%crucifixion%'
    """).fetchone()[0]
    print(f"  Subject 'crucifixion': {count:,} artworks")

    # Rembrandt as depicted person
    subject_fid = cur.execute("SELECT id FROM field_lookup WHERE name = 'subject'").fetchone()
    if subject_fid:
        count = cur.execute("""
            SELECT COUNT(DISTINCT m.artwork_id)
            FROM mappings m
            JOIN vocabulary v ON m.vocab_rowid = v.vocab_int_id
            WHERE m.field_id = ? AND v.type = 'person'
              AND v.label_en LIKE '%Rembrandt%'
        """, (subject_fid[0],)).fetchone()[0]
    else:
        count = 0
    print(f"  Depicted person 'Rembrandt': {count:,} artworks")

    # Amsterdam
    count = cur.execute("""
        SELECT COUNT(DISTINCT m.artwork_id)
        FROM mappings m
        JOIN vocabulary v ON m.vocab_rowid = v.vocab_int_id
        WHERE v.type = 'place' AND v.label_en LIKE '%Amsterdam%'
    """).fetchone()[0]
    print(f"  Place 'Amsterdam': {count:,} artworks")

    print("\n--- Top 10 Subjects ---")
    if subject_fid:
        rows = cur.execute("""
            SELECT v.notation, v.label_en, v.label_nl, v.type,
                   COUNT(DISTINCT m.artwork_id) as cnt
            FROM mappings m
            JOIN vocabulary v ON m.vocab_rowid = v.vocab_int_id
            WHERE m.field_id = ?
            GROUP BY m.vocab_rowid
            ORDER BY cnt DESC
            LIMIT 10
        """, (subject_fid[0],)).fetchall()
    else:
        rows = []
    for notation, en, nl, vtype, cnt in rows:
        label = en or nl or "?"
        code = notation if vtype == "classification" else "—"
        print(f"  {code or '—':12s} {label:50s} {cnt:6,}")

    # ── Importance score ─────────────────────────────────────────────
    # Delegates to compute-importance.py's shared function to avoid formula duplication.
    print("\n--- Importance Score ---")
    artworks_cols = get_columns(conn, "artworks")
    if "importance" not in artworks_cols:
        conn.execute("ALTER TABLE artworks ADD COLUMN importance INTEGER DEFAULT 0")
        conn.commit()
    from compute_importance import compute_importance_scores
    result = compute_importance_scores(conn, cur)
    for score, cnt in result["distribution"]:
        pct = cnt / result["total"] * 100
        print(f"  {score:3d}: {cnt:8,} ({pct:5.1f}%)")
    print(f"  Computed in {result['elapsed']:.1f}s")

    # Write version_info
    print("\n--- version_info ---")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS version_info (
            key TEXT PRIMARY KEY,
            value TEXT
        )
    """)
    built_at = datetime.now(timezone.utc).isoformat()
    artwork_count = cur.execute("SELECT COUNT(*) FROM artworks").fetchone()[0]
    vocab_count = cur.execute("SELECT COUNT(*) FROM vocabulary").fetchone()[0]
    mapping_count = cur.execute("SELECT COUNT(*) FROM mappings").fetchone()[0]
    version_rows = [
        ("built_at", built_at),
        ("artwork_count", str(artwork_count)),
        ("vocab_count", str(vocab_count)),
        ("mapping_count", str(mapping_count)),
    ]
    conn.executemany(
        "INSERT OR REPLACE INTO version_info (key, value) VALUES (?, ?)",
        version_rows,
    )
    conn.commit()
    for k, v in version_rows:
        print(f"  {k}: {v}")

    # Final VACUUM to reclaim space from dropped tables/columns
    print("\n--- VACUUM ---")
    t0 = time.time()
    conn.execute("VACUUM")
    print(f"  VACUUM complete in {time.time() - t0:.1f}s")

    if audit_results is not None:
        phase3_audit = run_phase_audit(conn, "phase3")
        audit_results["phase3"] = phase3_audit
        format_stdout_table(phase3_audit, "phase3")


# ─── Main ────────────────────────────────────────────────────────────

def parse_args() -> argparse.Namespace:
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Build the Rijksmuseum vocabulary SQLite database."
    )
    parser.add_argument(
        "--resume", action="store_true",
        help="Resume OAI-PMH harvest from last checkpoint",
    )
    parser.add_argument(
        "--skip-dump", action="store_true",
        help="Skip Phase 0 (useful when no local data dumps are available)",
    )
    parser.add_argument(
        "--phase", type=int, default=0, dest="start_phase",
        help="Run only from phase N onward (default: 0). Phase 4 = Tier 2 Linked Art resolution.",
    )
    parser.add_argument(
        "--threads", type=int, default=DEFAULT_THREADS,
        help=f"Thread count for Phase 4 Linked Art resolution (default: {DEFAULT_THREADS})",
    )
    parser.add_argument(
        "--geo-csv", type=str, default=None, dest="geo_csv",
        help="Path to geocoded places CSV (id,place_name,label_en,label_nl,external_id,lat,lon,artwork_count). "
             "Imports geocoding data into vocabulary table during Phase 3.",
    )
    parser.add_argument(
        "--limit", type=int, default=None, dest="limit_pages",
        help="Limit Phase 1 OAI-PMH harvest to N pages (~200 records/page). "
             "Default: unlimited. Subsequent phases auto-scope to harvested data.",
    )
    parser.add_argument(
        "--db", type=str, default=None,
        help="Override output database path (default: data/vocabulary.db).",
    )
    parser.add_argument(
        "--strict-audit", action="store_true",
        help="Exit with non-zero status at end of run if any audit target fails (default: warn only). See #222.",
    )
    parser.add_argument(
        "--stop-on-orphans", action="store_true",
        help="Halt before Phase 3 if the orphan vocab audit finds any live (non-KNOWN_DEAD) "
             "orphans. Forces a manual review of data/audit/orphan-vocab-ids-v0.24.csv before "
             "Phase 3's integer-encoding silently drops them. See #228.",
    )
    parser.add_argument(
        "--skip-enrichment", action="store_true",
        help="Skip the Phase 3 vocab enrichment step (person life dates, wikidata, place/concept hierarchy). "
             "DB will still open — column stubs are added unconditionally — but enriched fields will be empty. "
             "Useful for fast iteration when you don't need the offline dumps. See #242.",
    )
    parser.add_argument(
        "--dumps-dir", type=str, default=None, dest="dumps_dir",
        help=f"Override the default enrichment dumps directory (default: {DUMPS_DIR}). "
             "Expected to contain 201911-rma-edm-actors.zip, 201911-rma-skos-thesaurus.zip, "
             "place_extracted/, and person_extracted/. Missing artefacts are logged and skipped.",
    )
    return parser.parse_args()


def main():
    global DB_PATH, HARVEST_STARTED_AT
    args = parse_args()

    # #252: line-buffer stdout so `python harvest-vocabulary-db.py … | tee log`
    # surfaces progress in real time during multi-hour phases instead of
    # appearing dead between flushes. Equivalent to running with `python -u` or
    # PYTHONUNBUFFERED=1, but doesn't depend on the caller remembering it.
    sys.stdout.reconfigure(line_buffering=True)
    sys.stderr.reconfigure(line_buffering=True)

    if args.db:
        DB_PATH = Path(args.db)

    # #230: Stamp wall-clock start, repo-relative paths, and the harvest
    # script's git commit so the log + DB are self-describing.
    HARVEST_STARTED_AT = datetime.now(timezone.utc).isoformat()
    git_commit, git_dirty = detect_git_commit()
    git_label = f"{git_commit}{'-dirty' if git_dirty else ''}"

    print(f"Working dir: {PROJECT_DIR}")
    print(f"Harvest version: {HARVEST_VERSION}")
    print(f"Harvest commit: {git_label}")
    print(f"Started at: {HARVEST_STARTED_AT}")
    print(f"Database: {repo_path(DB_PATH)}")
    print(f"Options: resume={args.resume}, skip_dump={args.skip_dump}, start_phase={args.start_phase}, threads={args.threads}, geo_csv={args.geo_csv}, limit_pages={args.limit_pages}, skip_enrichment={args.skip_enrichment}, stop_on_orphans={args.stop_on_orphans}, dumps_dir={repo_path(args.dumps_dir or DUMPS_DIR)}")
    print()

    conn = create_or_open_db()

    # Audit accumulator (#222) — each phase appends its results; final_summary
    # writes the JSON artifact and (if --strict-audit) exits non-zero on FAIL.
    all_audit_results: dict[str, list[AuditResult]] = {}

    # Phase ordering: 0 → 0.5 → 1 → 2 → 4 → 2b → 3
    # Phase 2b re-resolves after Phase 4 (which introduces new vocab refs).
    # Phase 3 runs last because it builds FTS indexes and stats on all data.

    if args.start_phase <= 0 and not args.skip_dump:
        print("=== Phase 0: Parsing data dumps ===")
        t0 = time.time()
        run_phase0(conn)
        PHASE_DURATIONS["phase0"] = time.time() - t0
        print(f"  Phase 0 took {PHASE_DURATIONS['phase0']:.1f}s")
        phase0_audit = run_phase_audit(conn, "phase0")
        all_audit_results["phase0"] = phase0_audit
        format_stdout_table(phase0_audit, "phase0")
        print()

    if args.start_phase <= 0:
        print("=== Phase 0.5: Seeding curated set names ===")
        t0 = time.time()
        run_phase0_5(conn)
        PHASE_DURATIONS["phase0_5"] = time.time() - t0
        print(f"  Phase 0.5 took {PHASE_DURATIONS['phase0_5']:.1f}s")
        print()

    if args.start_phase <= 1:
        label = f"Harvesting OAI-PMH records (limit: {args.limit_pages} pages)" if args.limit_pages else "Harvesting ALL OAI-PMH records"
        print(f"=== Phase 1: {label} ===")
        t0 = time.time()
        run_phase1(conn, resume=args.resume, max_pages=args.limit_pages)
        PHASE_DURATIONS["phase1"] = time.time() - t0
        print(f"  Phase 1 took {PHASE_DURATIONS['phase1']:.1f}s")
        print()

    if args.start_phase <= 2:
        print("=== Phase 2: Resolving unmatched vocabulary URIs ===")
        t0 = time.time()
        run_phase2(conn)
        PHASE_DURATIONS["phase2"] = time.time() - t0
        print(f"  Phase 2 took {PHASE_DURATIONS['phase2']:.1f}s")
        phase2_audit = run_phase_audit(conn, "phase2")
        all_audit_results["phase2"] = phase2_audit
        format_stdout_table(phase2_audit, "phase2")
        print()

    if args.start_phase <= 4:
        print("=== Phase 4: Linked Art Resolution (Tier 2) ===")
        t0 = time.time()
        run_phase4(conn, threads=args.threads)
        PHASE_DURATIONS["phase4"] = time.time() - t0
        print(f"  Phase 4 took {PHASE_DURATIONS['phase4']:.1f}s")
        phase4_audit = run_phase_audit(conn, "phase4")
        all_audit_results["phase4"] = phase4_audit
        format_stdout_table(phase4_audit, "phase4")
        print()

        print("=== Phase 4.5: VisualItem.about[] thematic vocab (#283) ===")
        t0 = time.time()
        run_phase4_5(conn, threads=args.threads)
        PHASE_DURATIONS["phase4_5"] = time.time() - t0
        print(f"  Phase 4.5 took {PHASE_DURATIONS['phase4_5']:.1f}s")
        print()

        print("=== Phase 2b: Resolving new vocabulary URIs from Phase 4 ===")
        t0 = time.time()
        run_phase2(conn)
        PHASE_DURATIONS["phase2b"] = time.time() - t0
        print(f"  Phase 2b took {PHASE_DURATIONS['phase2b']:.1f}s")
        phase2b_audit = run_phase_audit(conn, "phase2")
        all_audit_results["phase2b"] = phase2b_audit
        format_stdout_table(phase2b_audit, "phase2b")
        print()

    # ── Orphan vocab audit (before Phase 3 integer-encoding drops them) ──
    print("=== Orphan Vocab Audit ===")
    orphan_sql = """
        SELECT m.vocab_id, m.field, COUNT(*) as cnt
        FROM mappings m
        LEFT JOIN vocabulary v ON m.vocab_id = v.id
        WHERE v.id IS NULL
        GROUP BY m.vocab_id, m.field
        ORDER BY cnt DESC
    """
    mapping_cols = get_columns(conn, "mappings")
    if "vocab_id" in mapping_cols:
        all_orphans = conn.execute(orphan_sql).fetchall()
        # #224: separate investigated-dead URIs so they don't retrigger pre-harvest review
        live_orphans = [(v, f, c) for v, f, c in all_orphans if v not in KNOWN_DEAD_URIS]
        dead_orphans = [(v, f, c) for v, f, c in all_orphans if v in KNOWN_DEAD_URIS]
        if dead_orphans:
            dead_mappings = sum(c for _, _, c in dead_orphans)
            print(f"  Known-dead URIs: {len(dead_orphans)} rows, {dead_mappings:,} mappings "
                  f"(excluded from orphan audit, see KNOWN_DEAD_URIS / #224)")
        if live_orphans:
            csv_path = PROJECT_DIR / "data" / "audit" / f"orphan-vocab-ids-v0.24.csv"
            csv_path.parent.mkdir(parents=True, exist_ok=True)
            with open(csv_path, "w") as f:
                f.write("vocab_id,field,count\n")
                for vid, field, cnt in live_orphans:
                    f.write(f"{vid},{field},{cnt}\n")
            print(f"  WARNING: {len(live_orphans)} orphan vocab IDs exported to {repo_path(csv_path)}")
            print(f"  Review and add missing codes to EXTERNAL_VOCAB before re-running Phase 3.")
            if args.stop_on_orphans:
                # #228: bail before Phase 3's integer-encoding drops these mappings.
                print()
                print(f"  --stop-on-orphans set: halting before Phase 3.")
                print(f"  Review {repo_path(csv_path)}, then resume with one of:")
                print(f"    a) Drop the orphans (accept silent loss): --phase 5")
                print(f"       (skips already-completed phases; runs orphan audit + Phase 3 only).")
                print(f"    b) Preserve some orphans: add their codes to EXTERNAL_VOCAB,")
                print(f"       then --phase 2 (re-runs Phase 2 → 4 → 2b → orphan audit → Phase 3).")
                conn.close()
                sys.exit(2)
        else:
            print(f"  No orphan vocab IDs found — all mappings have matching vocabulary entries.")
    else:
        print(f"  Skipping (mappings already integer-encoded)")
    print()

    print("=== Phase 3: Validation & Post-processing ===")
    dumps_dir_override = Path(args.dumps_dir) if args.dumps_dir else None
    t0 = time.time()
    run_phase3(
        conn,
        geo_csv=args.geo_csv,
        audit_results=all_audit_results,
        skip_enrichment=args.skip_enrichment,
        dumps_dir=dumps_dir_override,
    )
    PHASE_DURATIONS["phase3"] = time.time() - t0

    # #230: Stamp run-level provenance now that Phase 3 created version_info.
    # `built_at` was set inside run_phase3; we add finish timestamp, the harvest
    # script's git commit, and per-phase wall durations so the DB self-describes
    # which run produced it. Resumes (--start-phase N) only stamp the phases
    # that actually ran — keys for skipped phases are absent.
    finished_at = datetime.now(timezone.utc).isoformat()
    run_provenance: list[tuple[str, str]] = [
        ("harvest_started_at", HARVEST_STARTED_AT or ""),
        ("harvest_finished_at", finished_at),
        ("harvest_script_commit", git_commit),
        ("harvest_script_dirty", "1" if git_dirty else "0"),
        ("harvest_start_phase", str(args.start_phase)),
    ]
    for phase_key, secs in PHASE_DURATIONS.items():
        run_provenance.append((f"{phase_key}_duration_sec", f"{secs:.1f}"))
    conn.executemany(
        "INSERT OR REPLACE INTO version_info (key, value) VALUES (?, ?)",
        run_provenance,
    )
    conn.commit()
    print("\n--- run provenance (#230) ---")
    for k, v in run_provenance:
        print(f"  {k}: {v}")

    # Final audit summary (#222) — writes JSON artifact and exits non-zero on
    # FAIL when --strict-audit is set. Runs after run_phase3 so phase 3 drops
    # and final counts are reflected.
    final_summary(
        all_audit_results,
        strict_mode=args.strict_audit,
        version=HARVEST_VERSION,
    )

    conn.close()
    print(f"\nDone. Database at: {repo_path(DB_PATH)}")
    db_size = DB_PATH.stat().st_size / (1024 * 1024)
    print(f"Database size: {db_size:.1f} MB")
    print(f"Finished at: {finished_at}")


if __name__ == "__main__":
    main()
