Runtime¶

def build_context(
    root,
    options: dict | None = None,
    *,
    mode: str,
    xpath_env: XPathEnvironment | None = None,
    odd_namespaces: dict[str, str] | None = None,
    **settings,
) -> RenderContext:
    """The context a generated module's ``transform()`` runs with.

    *mode* names the output mode; its entry in [`opm.output_modes`][opm.output_modes]
    supplies the output functions and the text handling. *options* holds the
    ``$parameters`` plus the run options in `RUN_OPTIONS`. *xpath_env*
    supplies everything else XPath can see; an empty environment when omitted.
    *settings* are [`RenderContext`][opm.runtime.context.RenderContext] fields the module fixes: dispatch,
    stylesheet and the like.
    """
    from opm.output_modes import output_mode  # noqa: PLC0415

    from .xpath_env import XPathEnvironment  # noqa: PLC0415

    entry = output_mode(mode)
    opts = dict(options or {})
    env = xpath_env if xpath_env is not None else XPathEnvironment()
    parameters = {k: v for k, v in opts.items() if k not in RUN_OPTIONS}
    metadata = opts.get('metadata')
    return RenderContext(
        output=entry.name,
        pmf=entry.output_functions()(),
        parameters=parameters,
        xpath=env.for_odd(odd_namespaces).with_parameters(parameters),
        webcomponents=bool(opts.get('webcomponents')) and entry.webcomponents,
        docx_template=opts.get('docx_template'),
        input_path=opts.get('input_path'),
        root=root,
        # The caller's dict, when it passes one, so it can read what was collected.
        state=RunState(metadata=metadata if isinstance(metadata, dict) else {}),
        **entry.context_settings(),
        **settings,
    )

def current_environment() -> XPathEnvironment | None:
    """The environment evaluating the expression that is running right now.

    For functions that need the run's caches while XPath runs, such as
    ``fn:id()`` and ``tp:source-node``. ``None`` outside an evaluation.
    """
    return _CURRENT.get()

def params_cache_key(params: dict | None) -> tuple[tuple[str, str], ...]:
    """Hashable, order-independent form of a ``$parameters`` dict."""
    if not params:
        return ()
    return tuple(sorted((str(k), str(v)) for k, v in params.items()))

def default_element_namespace_uri(node: etree._Element) -> str:
    """Namespace URI used for unprefixed element names in XPath.

    XPath 3.1 binds unqualified names to this URI; TEI ``div`` lives in
    ``http://www.tei-c.org/ns/1.0``, not the empty namespace, so ``parent::div``
    only matches after setting this. Empty string means no default.
    """
    if isinstance(node, (etree._Comment, etree._ProcessingInstruction, etree._Entity)):
        return ''
    return etree.QName(node).namespace or ''

@lru_cache(maxsize=8192)
def compiled_xpath(
    expr: str,
    default_element_ns: str = '',
    ext_fp: str = '',
    namespaces: frozenset[tuple[str, str]] | None = None,
    base_uri: str | None = None,
):
    """Parse each distinct expression and static context once."""
    callables = dict(_BUILTIN_XPATH_CALLABLES)
    if ext_fp:
        # A project extension of the same name wins over the built-in.
        callables.update(_loaded_extension_callables(ext_fp))
    parser = build_extension_parser(
        default_element_ns,
        callables,
        namespaces=dict(namespaces) if namespaces else {},
        base_uri=base_uri,
    )
    return parser.parse(expr)

def normalize_extensions(extensions) -> tuple[str, ...]:
    """Extension modules as a tuple of non-empty dotted paths."""
    if not extensions:
        return ()
    if isinstance(extensions, str):
        extensions = (extensions,)
    return tuple(m for m in (str(mod).strip() for mod in extensions) if m)

def extension_fingerprint(extensions: tuple[str, ...]) -> str:
    """Cache-key fragment for *extensions*: module paths plus source mtimes."""
    if not extensions:
        return ''
    # The separator does not appear in module paths.
    return '\x1f'.join(fingerprint_for_module(module) for module in extensions)

def clear_compiled_xpath_cache() -> None:
    """Drop parsed expressions and loaded extension modules (e.g. between tests)."""
    compiled_xpath.cache_clear()
    _loaded_extension_callables.cache_clear()
    _parameters_map.cache_clear()

@contextmanager
def collect_xpath_errors() -> Iterator[XPathErrorLog]:
    """Record the XPath errors raised inside the ``with`` block."""
    log = XPathErrorLog()
    token = _ACTIVE.set(log)
    try:
        yield log
    finally:
        _ACTIVE.reset(token)

def record_xpath_error(
    expression: str,
    exc: Exception,
    node: etree._Element | None = None,
    base_uri: str | None = None,
) -> None:
    """Hand one error to the active log; does nothing outside a collection block."""
    log = _ACTIVE.get()
    if log is not None:
        log.record(expression, exc, node, base_uri)

def tag(node: etree._Element) -> str:
    """Local name for *node*.

    lxml comments, PIs, and entities use a Cython factory object as ``.tag``, not a
    string, so `etree.QName` cannot be used on them directly.
    """
    if isinstance(node, etree._Comment):
        return 'comment'
    if isinstance(node, etree._ProcessingInstruction):
        return 'processing-instruction'
    if isinstance(node, etree._Entity):
        return 'entity'
    return etree.QName(node).localname

def template_config(config):
    """*config* as seen inside a ``pb:template``.

    A behaviour combined with a ``pb:template`` receives the already-rendered
    template nodes as its content, so [`apply`][opm.runtime.pm_runtime.apply] and `apply_children`
    must hand them straight on instead of dispatching them again. Mirrors
    ``map:entry("template", true())`` in ``model.xql``, which is what stops
    tei-publisher-lib from reprocessing template output.

    Without it an ODD whose ``schemaSpec`` has ``ns=""`` (JATS, and any other
    vocabulary in no namespace) loses every element a template builds: the
    generated ``_dispatch`` passes foreign-namespace nodes through untouched,
    but for those ODDs the template's ``<li>`` looks exactly like a source
    element and falls through to "apply children", dropping the wrapper.
    """
    return config.derive(template=True)

def apply(config, nodes, dispatch):
    """Transform nodes via *dispatch(config, node, params)*."""
    if config.template:
        # Template output is finished markup — see [`template_config`][opm.runtime.pm_runtime.template_config].
        return list(normalize(nodes))
    params = config.parameters
    norm = config.normalize_text
    text_escape = config.text_escape
    result = []
    for node in nodes:
        if isinstance(node, (str, etree._ElementUnicodeResult)):
            text = maybe_normalize_text(str(node), norm)
            if text_escape and not isinstance(node, TemplateOutput):
                text = text_escape(text)
            result.append(text)
        elif isinstance(node, etree._Element) and not callable(node.tag):
            result.extend(dispatch(config, node, params))
    return result

def apply_template_param_value(config, source_node, raw):
    """Normalize and dispatch *raw* for ``pb:template`` ``[[param]]`` substitution.

    XPath (or a literal ``.`` param) may yield the context element itself. Passing
    that element through [`apply`][opm.runtime.pm_runtime.apply] would re-dispatch the same TEI node and, in
    templates, often stringifies it. When an item **is** *source_node*, recurse on
    ``child_nodes(source_node)`` instead (same rule as `apply_children`).
    """
    dispatch = config.dispatch
    norm = config.normalize_text
    text_escape = config.text_escape
    result = []
    for item in normalize(raw):
        if isinstance(item, (str, etree._ElementUnicodeResult)):
            text = maybe_normalize_text(str(item), norm)
            if text_escape and not isinstance(item, TemplateOutput):
                text = text_escape(text)
            result.append(text)
        elif isinstance(item, etree._Element):
            if item is source_node:
                result.extend(apply(config, child_nodes(source_node), dispatch))
            else:
                result.extend(apply(config, [item], dispatch))
        else:
            result.append(str(item))
    return result

def inject_cached_footnotes(nodes: list, config) -> list:
    """Append the footnote bodies collected in ``config.state`` after the main flow.

    HTML: [`HtmlOutputFunctions`][opm.runtime.html_output_functions.HtmlOutputFunctions] stores
    ``dl.footnote`` elements. Markdown: stores reference-definition strings.
    """
    footnotes = config.state.footnotes
    if not footnotes:
        return nodes
    if isinstance(footnotes[0], str):
        out = list(nodes) + list(footnotes)
        footnotes.clear()
        return out
    roots = [x for x in nodes if isinstance(x, etree._Element)]
    if not roots:
        return nodes
    target = _footnote_injection_target(roots)
    if target is None:
        return nodes
    for dl in footnotes:
        target.append(dl)
    footnotes.clear()
    return nodes

def expect_element(value: Any, *, arg_name: str = 'argument') -> ET._Element:
    """Unwrap an XPath item and require an lxml element.

    Extension functions receive XPath values; node items can arrive as
    `XPathNode` wrappers.
    """
    if isinstance(value, XPathNode):
        value = value.value
    if not isinstance(value, ET._Element):
        raise ValueError(f'{arg_name} expects an element node')
    return cast(ET._Element, value)

def expect_string(value: Any, *, arg_name: str = 'argument') -> str:
    """Normalize an XPath argument to a string.

    Accepts atomics, single-item sequences, XPathNode wrappers, and elements.
    Elements are converted from their string value (concatenated descendant text).
    """
    value = _unwrap_xpath_singleton(value)
    if isinstance(value, ET._Element):
        return ''.join(value.itertext())
    if value is None:
        raise ValueError(f'{arg_name} expects a string-compatible value')
    return str(value)

def expect_text(value: Any, *, arg_name: str = 'argument', strip: bool = True) -> str:
    """Like [`expect_string`][opm.runtime.xpath_extensions.expect_string] but intended for human-facing text."""
    text = expect_string(value, arg_name=arg_name)
    return text.strip() if strip else text

def fingerprint_for_module(module_dotted_path: str) -> str:
    """Cache key fragment: import path plus source mtime when available."""
    mod = importlib.import_module(module_dotted_path)
    path = getattr(mod, '__file__', None)
    if path:
        try:
            st = os.stat(path)
            return f'{module_dotted_path}\0{st.st_mtime_ns}'
        except OSError:
            pass
    return f'{module_dotted_path}\0'

def load_extension_callables(module_dotted_path: str) -> dict[str, Callable[..., Any]]:
    """Import *module_dotted_path* and collect public callables (name does not start with ``_``).

    Skips classes and non-routine callables so ``tp:`` functions map to plain functions/methods.
    """
    mod = importlib.import_module(module_dotted_path)
    out: dict[str, Callable[..., Any]] = {}
    for name in dir(mod):
        if name.startswith('_'):
            continue
        obj = getattr(mod, name)
        if inspect.isclass(obj):
            continue
        if not (inspect.isroutine(obj) or callable(obj)):
            continue
        # Unbound methods etc. are routines; avoid modules
        if inspect.ismodule(obj):
            continue
        out[name] = obj
    return out

def build_extension_parser(
    default_element_ns: str,
    callables: dict[str, Callable[..., Any]],
    namespaces: dict[str, str] | None = None,
    base_uri: str | None = None,
) -> OpmXPathParser:
    """Create an `OpmXPathParser` with ``tp:`` external functions."""
    ns = extension_namespace_map()
    if namespaces:
        ns = {**namespaces, **ns}  # ODD namespaces take precedence over tp: prefix
    kwargs: dict[str, Any] = {'namespaces': ns}
    if default_element_ns:
        kwargs['default_namespace'] = default_element_ns
    if base_uri:
        kwargs['base_uri'] = base_uri
    parser = OpmXPathParser(**kwargs)
    for name, fn in sorted(callables.items()):
        try:
            parser.external_function(
                fn,
                name=name,
                prefix=TEI_PUBLISHER_XPATH_EXT_PREFIX,
                sequence_types=(),
            )
        except ElementPathValueError as e:
            raise ElementPathValueError(
                f'XPath extension {name!r} could not be registered: {e}',
            ) from e
    return parser

def format_date(when: Any, locale: Any = 'en') -> str:
    """Format a TEI-style xs:date value for display in a popover/title."""
    when = expect_string(when, arg_name='format_date(when)')
    locale = expect_string(locale, arg_name='format_date(locale)')

    if re.fullmatch(r'\d{4}', when):
        return when

    if re.fullmatch(r'\d{4}-\d{2}', when):
        year_str, month_str = when.split('-', 1)
        try:
            parsed = date(int(year_str), int(month_str), 1)
        except ValueError:
            return when
        return babel_format_date(parsed, format='MMMM y', locale=locale)

    if re.fullmatch(r'\d{4}-\d{2}-\d{2}', when):
        try:
            parsed = datetime.strptime(when, '%Y-%m-%d').date()
        except ValueError:
            return when
        return babel_format_date(parsed, format='d MMMM y', locale=locale)

    return when

def heading_number(div: Any) -> str:
    """Port of ``pmf:heading-number`` from ``ext-common.xql`` (TEI ``div`` outline numbering).

    Returns a dotted index such as ``1.2.3``: at each level, the 1-based index among
    preceding ``tei:div`` siblings, joined from outer ancestor ``div`` down to *div*.
    """
    if isinstance(div, (list, tuple)):
        if len(div) != 1:
            raise ValueError('heading_number() expects a single node')
        div = div[0]
    node = expect_element(div, arg_name='heading_number()')

    parts: list[str] = []
    cur: ET._Element = node
    while True:
        n = 1
        for sib in cur.itersiblings(preceding=True):
            if _is_tei_div(sib):
                n += 1
        parts.append(str(n))
        parent = cur.getparent()
        if not _is_tei_div(parent):
            break
        cur = parent
    parts.reverse()
    return '.'.join(parts)