meshcore-bot/modules/response_template.py

#!/usr/bin/env python3
"""Piped placeholders for command response templates (feed-style ``{field|filter:args}``).

Used by :class:`~modules.commands.test_command.TestCommand` and extensible for other
commands. A placeholder holds either a bare field name (``{sender}``) or a
double-quoted string literal (``{"Hello {sender}!"}``); a quoted literal may embed
further ``{...}`` placeholders, which are expanded first and substituted into the
literal before any filters run. Either form may be followed by a ``|filter:arg``
chain, evaluated left to right.
"""

from __future__ import annotations

import asyncio
from typing import Any, Callable

from .url_shortener import shorten_url
from .utils import message_hop_count, message_path_bytes_per_hop

FilterFn = Callable[[str, dict[str, Any], str], str]


def _filter_pathbytes_min(value: str, ctx: dict[str, Any], args: str) -> str:
    """Clear *value* unless message path uses at least *N* bytes per hop (N in 1..3)."""
    message = ctx.get('message')
    if message is None:
        return ''
    try:
        n = int(args.strip())
    except ValueError:
        return value
    if n < 1 or n > 3:
        return value
    prefix_hex = int(ctx.get('prefix_hex_chars') or 2)
    bph = message_path_bytes_per_hop(message, prefix_hex_chars=prefix_hex)
    if bph < n:
        return ''
    return value


def _filter_hops_min(value: str, ctx: dict[str, Any], args: str) -> str:
    """Clear *value* unless the message travelled at least *N* hops.

    Asks about the route rather than how it is encoded, which is what separates
    this from ``pathbytes_min``: a one-byte multi-hop path has a real, measurable
    distance, and ``pathbytes_min:2`` would throw it away along with the direct
    messages it was aimed at. ``hops_min:1`` is the way to drop a clause on a
    direct message and nothing else.

    An unknown hop count clears the value: a gate that cannot confirm the route
    should suppress rather than guess, matching ``pathbytes_min``.
    """
    message = ctx.get('message')
    if message is None:
        return ''
    try:
        n = int(args.strip())
    except ValueError:
        return value
    if n < 0:
        return value
    hops = message_hop_count(message)
    if hops is None or hops < n:
        return ''
    return value


def _filter_prefix_if_nonempty(value: str, ctx: dict[str, Any], args: str) -> str:
    """Prepend *args* literal to *value* only when *value* is non-empty after prior filters."""
    if not value:
        return ''
    return args + value


def _filter_shorten_url(value: str, ctx: dict[str, Any], args: str) -> str:
    """Swap *value* for its shortened form, resolved ahead of time.

    Rendering is synchronous and runs on the event loop, so this filter never
    performs the HTTP request itself: a 5 s shortener timeout here would stall the
    radio transport along with everything else. :func:`resolve_template_async`
    does the network work off-thread first and leaves the answers in ``ctx``.

    On a miss the clause is dropped rather than falling back to the long URL. A
    v.gd link is ~19 bytes against a 158-160 byte message budget where a real
    analyzer URL is ~59, and ``_send_path_response`` subtracts the prefix from the
    first segment's budget — so falling back would quietly turn one transmission
    into two every time the shortener was unreachable.
    """
    if not value:
        return ''
    resolved = ctx.get('shortened')

    # Collection pass: record what needs shortening, change nothing.
    if isinstance(resolved, set):
        resolved.add(value)
        return value

    if isinstance(resolved, dict):
        return resolved.get(value, '')

    logger = ctx.get('logger')
    if logger is not None:
        logger.warning(
            "shorten_url used in a template rendered without resolve_template_async(); "
            "dropping the clause rather than blocking the event loop"
        )
    return ''


def _filter_if_nonempty(value: str, ctx: dict[str, Any], args: str) -> str:
    """Return *args* literal only when *value* is non-empty after prior filters."""
    if not value:
        return ''
    return args


RESPONSE_TEMPLATE_FILTERS: dict[str, FilterFn] = {
    'pathbytes_min': _filter_pathbytes_min,
    'pathbytes': _filter_pathbytes_min,
    'hops_min': _filter_hops_min,
    'prefix_if_nonempty': _filter_prefix_if_nonempty,
    'if_nonempty': _filter_if_nonempty,
    'shorten_url': _filter_shorten_url,
}

# prefix_if_nonempty's literal argument may itself contain '|', so once the parser
# sees this filter name it stops splitting on '|' and takes everything up to the
# placeholder's closing '}' as one literal argument. It must therefore be last in
# a chain whenever its literal needs a pipe.
_GREEDY_ARG_FILTERS = frozenset({'prefix_if_nonempty'})


class _TemplateParser:
    """Finite-state parser for ``{field|filter:arg|...}``-style placeholders.

    Walks the template left to right, character by character, alternating between
    plain text and placeholder spans. A placeholder's base value is either a bare
    field name or a ``"..."`` string literal; a literal may contain nested
    ``{...}`` placeholders (parsed recursively, same grammar) which are expanded
    before the literal is used as the base value for any following filters.
    """

    def __init__(self, template: str, fields: dict[str, Any], ctx: dict[str, Any], logger: Any):
        self.s = template
        self.n = len(template)
        self.fields = fields
        self.ctx = ctx
        self.logger = logger

    def render(self) -> str:
        out: list[str] = []
        i = 0
        while i < self.n:
            j = self.s.find('{', i)
            if j == -1:
                out.append(self.s[i:])
                break
            out.append(self.s[i:j])
            value, end = self._parse_placeholder(j)
            if value is None:
                out.append('{')
                i = j + 1
            else:
                out.append(value)
                i = end
        return ''.join(out)

    def _skip_ws(self, i: int) -> int:
        while i < self.n and self.s[i].isspace():
            i += 1
        return i

    def _parse_placeholder(self, start: int) -> tuple[str | None, int]:
        """Parse the placeholder beginning at ``self.s[start] == '{'``.

        Returns ``(expanded_value, index_after_closing_brace)``, or ``(None, start)``
        if there is no well-formed placeholder here (left as literal text).
        """
        i = start + 1
        if i < self.n and self.s[i] == '}':
            return None, start  # `{}` has no content
        i = self._skip_ws(i)
        if i >= self.n:
            return None, start
        if self.s[i] == '"':
            value, i = self._parse_quoted_string(i)
        else:
            value, i = self._parse_field_name(i)
        if value is None:
            return None, start

        i = self._skip_ws(i)
        filter_specs: list[tuple[str, str]] = []
        while i < self.n and self.s[i] == '|':
            i += 1
            name, args, i = self._parse_filter_spec(i)
            if name is None:
                return None, start
            filter_specs.append((name, args))
            i = self._skip_ws(i)

        if i >= self.n or self.s[i] != '}':
            return None, start
        raw_inner = self.s[start + 1:i]
        for name, args in filter_specs:
            value = self._apply_filter(name, args, value, raw_inner)
        return value, i + 1

    def _parse_field_name(self, i: int) -> tuple[str | None, int]:
        start = i
        while i < self.n and self.s[i] not in '|}':
            i += 1
        if i >= self.n:
            return None, i
        name = self.s[start:i].strip()
        return str(self.fields.get(name, '')), i

    def _parse_quoted_string(self, i: int) -> tuple[str | None, int]:
        """Parse a ``"..."`` literal starting at the opening quote.

        ``\\"`` and ``\\\\`` are recognized escapes; any ``{...}`` inside the
        literal is expanded recursively and substituted in place.
        """
        i += 1
        parts: list[str] = []
        while i < self.n:
            ch = self.s[i]
            if ch == '\\' and i + 1 < self.n and self.s[i + 1] in ('"', '\\'):
                parts.append(self.s[i + 1])
                i += 2
                continue
            if ch == '"':
                return ''.join(parts), i + 1
            if ch == '{':
                value, i = self._parse_placeholder(i)
                if value is None:
                    return None, i
                parts.append(value)
                continue
            parts.append(ch)
            i += 1
        return None, i  # unterminated string literal

    def _parse_filter_spec(self, i: int) -> tuple[str | None, str, int]:
        start = i
        while i < self.n and self.s[i] not in ':|}':
            i += 1
        if i >= self.n:
            return None, '', i
        name = self.s[start:i].strip()
        if i >= self.n or self.s[i] != ':':
            return name, '', i
        i += 1  # consume ':'
        if name in _GREEDY_ARG_FILTERS:
            close = self.s.find('}', i)
            if close == -1:
                return None, '', self.n
            return name, self.s[i:close], close
        quoted_at = self._skip_ws(i)
        if quoted_at < self.n and self.s[quoted_at] == '"':
            value, j = self._parse_quoted_string(quoted_at)
            if value is None:
                return None, '', j
            return name, value, j
        arg_start = i
        while i < self.n and self.s[i] not in '|}':
            i += 1
        return name, self.s[arg_start:i], i

    def _apply_filter(self, name: str, args: str, value: str, raw_inner: str) -> str:
        fn = RESPONSE_TEMPLATE_FILTERS.get(name)
        if fn is None:
            if self.logger is not None:
                self.logger.warning(f"Unknown response template filter {name!r} in {{{raw_inner}}}")
            return value
        return fn(value, self.ctx, args)


def format_piped_template(
    template: str,
    fields: dict[str, Any],
    *,
    message: Any = None,
    logger: Any = None,
    config: Any = None,
    shortened: dict[str, str] | None = None,
    prefix_hex_chars: int = 2,
) -> str:
    """Replace ``{field}``, ``{"literal {field}"}``, and their piped filter chains.

    Args:
        template: Raw template string from config.
        fields: Mapping of placeholder names to values (e.g. ``sender``, ``path_distance``).
            An unavailable field is an empty string, which renders as nothing and
            lets ``prefix_if_nonempty`` drop its literal label too.
        message: Triggering mesh message; required for ``pathbytes`` / ``pathbytes_min`` filters.
        logger: Optional logger for unknown filter warnings.
        config: Bot config, for filters that read ``[External_Data]``.
        shortened: Long-URL to short-URL mapping from :func:`resolve_template_async`.
            Required by the ``shorten_url`` filter, which will not make a network
            call from this synchronous path.
        prefix_hex_chars: Bot prefix width for inferring bytes per hop from legacy path text.

    Returns:
        Fully expanded string.
    """
    ctx: dict[str, Any] = {
        'message': message,
        'logger': logger,
        'prefix_hex_chars': prefix_hex_chars,
        'config': config,
        'shortened': shortened,
    }
    if logger is not None:
        logger.debug("Rendering response template %r with fields %r", template, fields)
    return _TemplateParser(template, fields, ctx, logger).render()


def template_needs_resolution(template: str) -> bool:
    """True if *template* uses a filter that needs :func:`resolve_template_async`.

    A cheap substring test so the common template pays nothing for a feature it
    does not use; the collection pass below is what actually decides.
    """
    return 'shorten_url' in template


async def resolve_template_async(
    template: str,
    fields: dict[str, Any],
    *,
    message: Any = None,
    logger: Any = None,
    config: Any = None,
    prefix_hex_chars: int = 2,
) -> dict[str, str]:
    """Resolve *template*'s network-backed filters off the event loop.

    Renders the template once with ``shorten_url`` in collection mode, which walks
    the real filter chain — so gating filters such as ``hops_min`` have already had
    their say and a suppressed clause costs no request — then shortens whatever
    survived, concurrently and in a worker thread. Pass the result to
    :func:`format_piped_template` as ``shortened``.

    Returns an empty mapping when there is nothing to do, which renders exactly as
    an unresolved template would.
    """
    if config is None or not template_needs_resolution(template):
        return {}

    pending: set[str] = set()
    ctx: dict[str, Any] = {
        'message': message,
        'logger': logger,
        'prefix_hex_chars': prefix_hex_chars,
        'config': config,
        'shortened': pending,
    }
    _TemplateParser(template, fields, ctx, logger).render()
    if not pending:
        return {}

    urls = sorted(pending)
    results = await asyncio.gather(
        *(shorten_url(u, config=config, logger=logger) for u in urls),
        return_exceptions=True,
    )
    resolved: dict[str, str] = {}
    for url, short in zip(urls, results, strict=True):
        if isinstance(short, BaseException):
            if logger is not None:
                logger.debug("Shortening %r failed: %s", url, short)
            continue
        if short:
            resolved[url] = short
    return resolved