Core¶

def coerce_slot_selector(payload: Any) -> str | None:
    """Normalise a selector definition coming from front matter."""
    if isinstance(payload, str):
        candidate = payload.strip()
        return candidate or None
    if isinstance(payload, Mapping):
        for key in ("label", "title", "section"):
            value = payload.get(key)
            if isinstance(value, str) and value.strip():
                return value.strip()
    return None

def extract_front_matter_slots(
    front_matter: Mapping[str, Any],
) -> tuple[dict[str, str], dict[str, SlotOptions]]:
    """Return front-matter slot selectors alongside their per-slot options."""
    root_slots = front_matter.get("slots") or front_matter.get("entrypoints")
    return parse_slot_mapping(root_slots)

def parse_slot_mapping(
    raw: Any,
) -> tuple[dict[str, str], dict[str, SlotOptions]]:
    """Parse slot mappings together with their per-slot options.

    Accepts the three front-matter shapes ``{name: selector}``,
    ``[{target/slot: ..., label: ...}]`` and the bare ``"name:selector"``
    string form. Returns a ``(selectors, options)`` tuple where ``options``
    only contains entries whose values differ from the ``SlotOptions``
    defaults, so callers can ignore it when they don't care.
    """
    overrides: dict[str, str] = {}
    options: dict[str, SlotOptions] = {}
    if not raw:
        return overrides, options

    def _record(name: str, selector: str, option_payload: Any) -> None:
        key = name.strip()
        if not key or not selector:
            return
        overrides[key] = selector
        slot_options = _extract_slot_options(option_payload)
        if slot_options != SlotOptions():
            options[key] = slot_options

    if isinstance(raw, Mapping):
        for slot_name, payload in raw.items():
            if not isinstance(slot_name, str):
                continue
            selector = coerce_slot_selector(payload)
            if selector:
                _record(slot_name, selector, payload)
        return overrides, options

    if isinstance(raw, Iterable) and not isinstance(raw, str | bytes):
        for entry in raw:
            if not isinstance(entry, Mapping):
                continue
            slot_name = entry.get("target") or entry.get("slot")
            if not isinstance(slot_name, str):
                continue
            selector = entry.get("label") or entry.get("title") or entry.get("section")
            selector_value = coerce_slot_selector(selector)
            if not selector_value:
                selector_value = coerce_slot_selector(entry)
            if selector_value:
                _record(slot_name, selector_value, entry)
        return overrides, options

    if isinstance(raw, str):
        entry = raw.strip()
        if entry and ":" in entry:
            name, selector = entry.split(":", 1)
            selector = selector.strip()
            if selector:
                _record(name, selector, None)
        return overrides, options

    return overrides, options

def debug_enabled(emitter: DiagnosticEmitter | None) -> bool:
    """Return whether debug mode is active for the given emitter."""
    return emitter is not None and emitter.debug_enabled

def default_severity(code: str) -> Severity:
    """The severity a code carries when the emitter names none.

    A code absent from the table (a tmark id not listed here) is a warning:
    tmark records arrive with their own severity anyway.
    """
    info = CODES.get(code)
    return info.severity if info is not None else Severity.WARNING

def emit_diagnostic(
    emitter: DiagnosticEmitter | None,
    code: str,
    message: str,
    *,
    span: Span = NO_SPAN,
    severity: Severity | None = None,
    exc: BaseException | None = None,
) -> None:
    """Record a coded finding through ``emitter``.

    The counterpart of :meth:`DiagnosticSink.emit` for the stages that hold an
    emitter rather than a :class:`~texsmith.passes.PassContext`. It goes
    through :meth:`DiagnosticEmitter.diagnostic` rather than reaching the sink
    directly, so an emitter that filters sees these records too; the severity
    defaults to the one the code table gives the code.
    """
    ensure_emitter(emitter).diagnostic(
        Diagnostic(
            code=code,
            severity=severity if severity is not None else default_severity(code),
            span=span,
            message=message,
        ),
        exc,
    )

def ensure_emitter(emitter: DiagnosticEmitter | None) -> DiagnosticEmitter:
    """Return a usable emitter, defaulting to the null implementation."""
    return emitter if emitter is not None else NullEmitter()

def format_diagnostic(diagnostic: Diagnostic, files: FileTable, *, verbosity: int = 0) -> str:
    """The ``tmark-cli`` line; the fix and the related locations indented at ``-v``.

    A multi-line message keeps its first line on the header and indents the
    rest, so the detail an emitter appends (``-v`` exception notes) stays
    attached to its record.
    """
    first, _, rest = diagnostic.message.partition("\n")
    lines = [
        f"{_location_prefix(diagnostic.span, files)}"
        f"{diagnostic.severity.value} {diagnostic.code}: {first}"
    ]
    lines.extend(f"  {line}" for line in rest.splitlines())
    if verbosity >= 1:
        if diagnostic.fix is not None:
            where = _location_prefix(diagnostic.fix.span, files).rstrip(": ")
            target = f" at {where}" if where else ""
            lines.append(f"  fix{target}: replace with {diagnostic.fix.replacement!r}")
        for span, label in diagnostic.related:
            lines.append(f"  {_location_prefix(span, files)}{label}")
    return "\n".join(lines)

def format_event_message(name: str, payload: Mapping[str, Any]) -> str | None:
    """Return a human-friendly summary for selected diagnostic events."""
    try:
        data = dict(payload)
    except Exception:  # pragma: no cover - defensive
        data = {}

    if name == "asset_fetch":
        url = data.get("url") or "<unknown>"
        convert = data.get("convert")
        suffix_hint = data.get("suffix_hint")
        details: list[str] = []
        if convert:
            details.append("convert")
        if suffix_hint:
            details.append(f"suffix={suffix_hint}")
        suffix = f" ({', '.join(details)})" if details else ""
        return f"Fetching: {url}{suffix}"

    if name == "asset_fetch_cached":
        url = data.get("url") or "<unknown>"
        reason = data.get("reason") or "cache"
        return f"Reusing cached remote image: {url} ({reason})"

    if name == "doi_fetch":
        doi_value = data.get("value") or data.get("doi") or "<unknown>"
        key = data.get("key") or "<unknown>"
        mode = data.get("mode")
        source = data.get("source") or data.get("resolved_source")
        details: list[str] = []
        if mode:
            details.append(str(mode))
        if source:
            details.append(str(source))
        suffix = f" ({', '.join(details)})" if details else ""
        return f"Resolved DOI {doi_value} for entry '{key}'{suffix}"

    return None

def from_tmark(record: Mapping[str, Any]) -> Diagnostic:
    """Build a :class:`Diagnostic` from a record returned by the tmark bindings.

    The record is the ``serde`` JSON of ``tmark_ir::Diagnostic``: ``fix`` and
    ``related`` are absent when empty. ``origin`` is forced to ``"tmark"``.
    """
    fix_payload = record.get("fix")
    related_payload = record.get("related") or ()
    return Diagnostic(
        code=str(record["code"]),
        severity=Severity(str(record["severity"])),
        span=Span.from_json(record["span"]),
        message=str(record["message"]),
        fix=Fix.from_json(fix_payload) if fix_payload is not None else None,
        related=tuple((Span.from_json(span), str(label)) for span, label in related_payload),
        origin="tmark",
    )

def locate(span: Span, files: FileTable) -> tuple[Any, int | None, int | None] | None:
    """``(path, line, col)`` of a span; line and col are ``None`` without a text.

    ``NO_SPAN`` names the main document without a position; an unknown file
    yields ``None`` and the record prints without a path.
    """
    entry = files.get(span.file)
    if entry is None:
        return None
    if span == NO_SPAN or not entry.text:
        return (entry.path, None, None)
    position = entry.line_index.line_col(span.start)
    return (entry.path, position.line, position.col)

def record_event(
    emitter: DiagnosticEmitter | None,
    event: str,
    payload: Mapping[str, Any],
) -> None:
    """Forward a structured diagnostic event."""
    ensure_emitter(emitter).event(event, payload)

def summary_line(counts: Mapping[Severity, int]) -> str:
    """``N errors, M warnings`` — singular when it applies."""
    errors = counts.get(Severity.ERROR, 0)
    warnings = counts.get(Severity.WARNING, 0)
    return (
        f"{errors} error{'' if errors == 1 else 's'}, "
        f"{warnings} warning{'' if warnings == 1 else 's'}"
    )

def unknown_container(diagnostic: Diagnostic) -> str | None:
    """The container name a ``container-unknown`` record names, else ``None``."""
    if diagnostic.code != CONTAINER_UNKNOWN:
        return None
    match = _CONTAINER_UNKNOWN_NAME.match(diagnostic.message)
    return match.group(1) if match is not None else None

def exception_hint(exc: BaseException) -> str | None:
    """Return the most specific message available for an exception chain."""
    messages = exception_messages(exc)
    return messages[-1] if messages else None

def exception_messages(exc: BaseException) -> list[str]:
    """Return the collected message chain for an exception and its causes."""
    messages: list[str] = []
    visited: set[int] = set()
    current: BaseException | None = exc
    while current is not None and id(current) not in visited:
        visited.add(id(current))
        text = str(current).strip()
        if text:
            first_line = text.splitlines()[0].strip()
            if first_line:
                messages.append(first_line)
        current = current.__cause__ or current.__context__
    return messages

def format_rendering_error(error: LatexRenderingError) -> str:
    """Format a human-readable rendering failure summary."""
    cause = error.__cause__
    if cause is None:
        return str(error)
    return f"LaTeX rendering failed: {cause}"

def format_user_friendly_render_error(error: LatexRenderingError) -> str:
    """Return a concise rendering failure summary suitable for end users."""
    summary = "LaTeX rendering failed"
    hint_source = error.__cause__ or error
    hint = exception_hint(hint_source)
    if hint:
        summary = f"{summary}: {hint}"
    if summary.endswith("."):
        summary = summary.rstrip(".")
    return f"{summary}. Re-run with --debug for technical details."

def raise_conversion_error(
    emitter: DiagnosticEmitter | None,
    message: str,
    exc: Exception,
) -> None:
    """Record the failure that stops the run, then raise it.

    Lives here rather than with the emitters: it builds a
    :class:`ConversionError`, and the diagnostics package must not depend on
    the conversion one.
    """
    emit_diagnostic(emitter, "conversion-failed", message, exc=exc)
    error = ConversionError(message)
    error._texsmith_logged = True  # noqa: SLF001
    raise error from exc

def escape_latex_chars(text: str, *, legacy_accents: bool = False) -> str:
    """Escape LaTeX special characters leveraging pylatexenc."""
    if not text:
        return text
    parts: list[str] = []
    buffer: list[str] = []

    def _encode_chunk(chunk: str) -> str:
        escaped = "".join(_BASIC_LATEX_ESCAPE_MAP.get(char, char) for char in chunk)
        if legacy_accents:
            encoded = unicode_to_latex(escaped, non_ascii_only=True, unknown_char_warning=False)
            return _wrap_latex_output(encoded)
        return escaped

    def _should_skip_encoding(char: str) -> bool:
        try:
            name = unicodedata.name(char)
        except ValueError:
            return False
        if "SUPERSCRIPT" in name or "SUBSCRIPT" in name:
            return True
        return "MODIFIER LETTER" in name and ("SMALL" in name or "CAPITAL" in name)

    for char in text:
        replacement = _COMMON_SYMBOL_MAP.get(char)
        if replacement is not None:
            if buffer:
                parts.append(_encode_chunk("".join(buffer)))
                buffer.clear()
            parts.append(replacement)
            continue
        if _should_skip_encoding(char):
            if buffer:
                parts.append(_encode_chunk("".join(buffer)))
                buffer.clear()
            parts.append(char)
        else:
            buffer.append(char)

    if buffer:
        parts.append(_encode_chunk("".join(buffer)))

    return "".join(parts)