LaTeX¶

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)

def escape_text_segment(
    text: str,
    *,
    legacy_accents: bool = False,
    emoji_renderer: Callable[[str], str] | None = None,
) -> str:
    """Escape a plain-text run, splitting out emoji clusters.

    Mirrors ``adapters/handlers/inline.py:_escape_text_segment``. Plain
    fragments go through :func:`prepare_plain_text`; emoji clusters are handed
    to ``emoji_renderer`` (font command or artifact image). When no renderer is
    supplied, emoji are escaped as ordinary text.
    """
    chunks: list[str] = []
    for kind, payload in _segment_text_with_emoji(text):
        if kind == "text":
            if payload:
                chunks.append(prepare_plain_text(payload, legacy_accents=legacy_accents))
        elif emoji_renderer is not None:
            chunks.append(emoji_renderer(payload))
        else:
            chunks.append(prepare_plain_text(payload, legacy_accents=legacy_accents))
    return "".join(chunks)

def prepare_plain_text(text: str, *, legacy_accents: bool = False) -> str:
    """Escape and apply the unicode text transforms for a plain-text run.

    Mirrors ``adapters/handlers/inline.py:_prepare_plain_text``: smart quotes
    and dashes are normalised first, the text is escaped, then unicode
    sub/superscript runs become ``\\textsubscript`` / ``\\textsuperscript``.
    """
    text = _replace_unicode_punctuation(text)
    text = _replace_unicode_dashes(text)
    escaped = escape_latex_chars(text, legacy_accents=legacy_accents)
    escaped = _replace_unicode_superscripts(escaped)
    return _replace_unicode_subscripts(escaped)

def store_local_image_asset(
    opts: AssetOptions,
    resolved: Path,
    *,
    options: Mapping[str, str] | None = None,
) -> Path:
    """Copy or convert a local asset and register it on the context.

    ``options`` are the per-image render options (``crop`` for draw.io) the
    document set through ``attr_list``. They select the converter behaviour and
    therefore take part in the asset key: the same diagram rendered cropped and
    uncropped is two distinct assets.
    """
    asset_key = _asset_key(resolved, options)
    existing = opts.assets.lookup(asset_key)
    if existing is not None:
        return existing

    suffix = resolved.suffix.lower()
    convert_requested = opts.convert_assets
    needs_conversion = _requires_conversion(suffix, convert_requested)
    emitter = ensure_emitter(opts.emitter)

    if needs_conversion:
        record_event(
            emitter,
            "asset_convert",
            {
                "source": str(resolved),
                "suffix": suffix,
                "reason": "requested" if convert_requested else "forced",
            },
        )
        staged = _convert_local_asset(opts, resolved, suffix, options)
        final_suffix = ".pdf"
    else:
        staged = resolved
        final_suffix = suffix or ".bin"

    record_event(
        emitter,
        "asset_local",
        {
            "source": str(resolved),
            "stored_suffix": final_suffix,
            "converted": needs_conversion,
        },
    )
    return _persist_asset(
        opts,
        asset_key=asset_key,
        staged_path=staged,
        suffix=final_suffix,
        source_path=resolved,
    )

def store_remote_image_asset(opts: AssetOptions, url: str) -> Path:
    """Fetch a remote asset, mirror it locally, and register it."""
    existing = opts.assets.lookup(url)
    if existing is not None:
        return existing

    convert_requested = opts.convert_assets
    metadata: dict[str, str] = {}
    conversion_root = _conversion_cache_root(opts)
    manifest_path = conversion_root / _ASSET_MANIFEST
    manifest, manifest_dirty = _load_asset_manifest(manifest_path)
    url_suffix = _suffix_from_url(url)
    suffix_hint = _normalise_suffix(url_suffix, default="") if url_suffix else ""
    emitter = ensure_emitter(opts.emitter)

    record_event(
        emitter,
        "asset_fetch",
        {
            "url": url,
            "convert": convert_requested,
            "suffix_hint": suffix_hint,
        },
    )
    fetch_options: dict[str, Any] = {
        "convert": convert_requested,
        "metadata": metadata,
        "manifest": manifest,
        "manifest_path": manifest_path,
        "manifest_dirty": manifest_dirty,
        "emitter": emitter,
    }
    user_agent = opts.http_user_agent
    if user_agent:
        fetch_options["user_agent"] = user_agent
    if convert_requested:
        fetch_options["output_suffix"] = ".pdf"
    elif suffix_hint:
        fetch_options["output_suffix"] = suffix_hint
    artefact = fetch_image(
        url,
        output_dir=conversion_root,
        **fetch_options,
    )

    recorded_suffix = metadata.get("suffix") if metadata else None
    if not recorded_suffix:
        detected_suffix = _detect_file_suffix(Path(artefact))
        if detected_suffix:
            recorded_suffix = detected_suffix
    if recorded_suffix:
        final_suffix = _normalise_suffix(recorded_suffix)
    else:
        final_suffix = Path(artefact).suffix or ".pdf"
    prefer_name = _extract_remote_name(url)

    record_event(
        emitter,
        "asset_fetch_complete",
        {
            "url": url,
            "stored_suffix": final_suffix,
            "converted": convert_requested or suffix_hint in _FORCED_CONVERSION_SUFFIXES,
        },
    )
    if manifest_dirty["dirty"]:
        _save_asset_manifest(manifest_path, manifest)
    return _persist_asset(
        opts,
        asset_key=url,
        staged_path=artefact,
        suffix=final_suffix,
        source_path=None,
        prefer_name=prefer_name,
        force_hash=not bool(prefer_name),
    )

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)

def parse_latex_log(log_path: Path) -> list[LatexMessage]:
    """Parse a LaTeX log file into structured messages."""
    if not log_path.exists():
        return []
    parser = LatexLogParser()
    with log_path.open("r", encoding="utf-8", errors="replace") as handle:
        for line in handle:
            parser.process_line(line)
    parser.finalize()
    return list(parser.messages)

def stream_latexmk_output(
    command: Sequence[str],
    *,
    cwd: str,
    env: Mapping[str, str],
    console: Console,
    verbosity: int = 0,
) -> LatexStreamResult:
    """Execute a LaTeX engine command and render output incrementally."""
    from .log_render import LatexLogRenderer

    parser = LatexLogParser()
    renderer = LatexLogRenderer(console)

    with subprocess.Popen(
        command,
        cwd=cwd,
        env=dict(env),
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        bufsize=1,
        encoding="utf-8",
        errors="replace",
    ) as process:
        selector = selectors.DefaultSelector()
        if process.stdout:
            selector.register(process.stdout, selectors.EVENT_READ)
        if process.stderr:
            selector.register(process.stderr, selectors.EVENT_READ)

        while selector.get_map():
            for key, _ in selector.select():
                stream_obj = key.fileobj
                if isinstance(stream_obj, int) or not hasattr(stream_obj, "readline"):
                    selector.unregister(stream_obj)
                    continue
                stream = cast(TextIO, stream_obj)
                chunk = stream.readline()
                if chunk:
                    for completed in parser.process_line(chunk):
                        if _should_emit_message(completed, verbosity):
                            renderer.consume(completed)
                else:
                    selector.unregister(stream)

        for completed in parser.finalize():
            if _should_emit_message(completed, verbosity):
                renderer.consume(completed)

        returncode = process.wait()

    renderer.summarize()
    return LatexStreamResult(returncode=returncode, messages=renderer.messages)

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)

def build_template_overrides(front_matter: Mapping[str, Any] | None) -> dict[str, Any]:
    """Build template overrides from front matter while preserving metadata."""
    if not front_matter or not isinstance(front_matter, Mapping):
        return {}

    overrides = dict(front_matter)
    press_section = overrides.get("press")
    if isinstance(press_section, Mapping):
        overrides["press"] = dict(press_section)

    fragments = overrides.get("fragments")
    if fragments is None and isinstance(press_section, Mapping):
        fragments = press_section.get("fragments")
    if isinstance(fragments, list):
        overrides["fragments"] = list(fragments)

    callouts_section = overrides.get("callouts")
    if callouts_section is None and isinstance(press_section, Mapping):
        callouts_section = press_section.get("callouts")
    if isinstance(callouts_section, Mapping):
        overrides["callouts"] = dict(callouts_section)

    callouts_style = overrides.get("callouts_style")
    if callouts_style is None and isinstance(press_section, Mapping):
        callouts_style = press_section.get("callouts_style")
    if callouts_style is not None:
        overrides["callout_style"] = callouts_style

    base_override = overrides.get("base_level")
    if base_override is None and isinstance(press_section, Mapping):
        base_override = press_section.get("base_level")
    if base_override is not None:
        try:
            overrides["base_level"] = coerce_base_level(base_override)
        except TemplateError:
            overrides["base_level"] = base_override

    return overrides

def coerce_base_level(value: Any, *, allow_none: bool = True) -> int | None:
    """Normalise base-level metadata to an integer or ``None``."""
    if value is None:
        if allow_none:
            return None
        raise TemplateError("Base level value is missing.")

    if isinstance(value, bool):
        raise TemplateError("Base level must be an integer, booleans are not supported.")

    if isinstance(value, (int, float)):
        return int(value)

    if isinstance(value, str):
        candidate = value.strip().lower()
        if not candidate:
            if allow_none:
                return None
            raise TemplateError("Base level value cannot be empty.")
        alias_map = {
            "part": -1,
            "chapter": 0,
            "section": 1,
            "subsection": 2,
        }
        if candidate in alias_map:
            return alias_map[candidate]
        try:
            return int(candidate)
        except ValueError as exc:  # pragma: no cover - defensive
            raise TemplateError(
                f"Invalid base level '{value}'. Expected an integer value or one of "
                f"{', '.join(alias_map)}."
            ) from exc

    raise TemplateError(
        f"Base level should be provided as an integer value, got type '{type(value).__name__}'."
    )

def copy_template_assets(
    template: WrappableTemplate,
    output_dir: Path,
    *,
    context: Mapping[str, Any] | None = None,
    overrides: Mapping[str, Any] | None = None,
    assets: Iterable[ResolvedAsset] | None = None,
) -> list[Path]:
    """Copy the template declared assets into the selected output directory."""
    output_dir = Path(output_dir).resolve()
    output_dir.mkdir(parents=True, exist_ok=True)

    if context is None:
        render_context = template.prepare_context("", overrides=overrides)
    else:
        render_context = dict(context)

    written: list[Path] = []
    asset_entries = list(assets) if assets is not None else list(template.iter_assets())
    for asset in asset_entries:
        destination_path = (output_dir / asset.destination).resolve()

        if asset.template:
            if asset.template_name is None:  # pragma: no cover - defensive
                raise TemplateError(
                    f"Templated asset '{asset.source}' is missing template metadata."
                )
            destination_path.parent.mkdir(parents=True, exist_ok=True)
            rendered = template.render_template(asset.template_name, **render_context)
            destination_path.write_text(
                rendered,
                encoding=asset.encoding or "utf-8",
            )
        elif asset.source.is_dir():
            shutil.copytree(asset.source, destination_path, dirs_exist_ok=True)
        else:
            destination_path.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(asset.source, destination_path)

        written.append(destination_path)

    return written

def declared_containers(template: str | None) -> frozenset[str]:
    """The container names ``template`` declares (``containers``), for both backends.

    Read straight from the manifest, because it has to be known *before* the
    documents are parsed: tmark reports ``::: solution`` as
    ``container-unknown`` while it parses, and no pass — not even the
    template's own, which rewrites that very container — can take a record
    back. The parse is one for both backends (``prepare_documents`` runs before
    the LaTeX binding is resolved and before the Typst template is loaded), so
    the names of ``[latex.template]`` and ``[typst.template]`` are read
    together.

    A template that cannot be loaded declares nothing: that failure belongs to
    the render, which reports it with its own message.
    """
    if not template:
        return frozenset()
    try:
        instance = load_template(template)
    except TemplateError:
        return frozenset()
    manifest = instance.manifest
    names = set(manifest.latex.template.containers)
    if manifest.typst is not None:
        names.update(manifest.typst.template.containers)
    return frozenset(names)