markdown.py

"""Define classes for Markdown serialization."""

import html
import logging
import re
import textwrap
from enum import Enum
from pathlib import Path
from typing import Annotated, Any, Optional, Union

from pydantic import AnyUrl, BaseModel, Field, PositiveInt, model_validator
from tabulate import _column_type, tabulate
from typing_extensions import override

from docling_core.transforms.serializer.base import (
    BaseAnnotationSerializer,
    BaseDocSerializer,
    BaseFallbackSerializer,
    BaseFormSerializer,
    BaseInlineSerializer,
    BaseKeyValueSerializer,
    BaseListSerializer,
    BaseMetaSerializer,
    BasePictureSerializer,
    BaseTableSerializer,
    BaseTextSerializer,
    SerializationResult,
)
from docling_core.transforms.serializer.common import (
    CommonParams,
    DocSerializer,
    _get_annotation_text,
    _should_use_legacy_annotations,
    create_ser_result,
)
from docling_core.types.doc.base import ImageRefMode
from docling_core.types.doc.document import (
    BaseMeta,
    CodeItem,
    ContentLayer,
    DescriptionAnnotation,
    DescriptionMetaField,
    DocItem,
    DocItemLabel,
    DoclingDocument,
    FloatingItem,
    Formatting,
    FormItem,
    FormulaItem,
    GroupItem,
    ImageRef,
    InlineGroup,
    KeyValueItem,
    KeywordsMetaField,
    ListGroup,
    ListItem,
    MoleculeMetaField,
    NodeItem,
    PictureClassificationData,
    PictureClassificationMetaField,
    PictureItem,
    PictureMoleculeData,
    PictureTabularChartData,
    RichTableCell,
    SectionHeaderItem,
    SummaryMetaField,
    TableItem,
    TabularChartMetaField,
    TextItem,
    TitleItem,
    TopicsMetaField,
)

_logger = logging.getLogger(__name__)


def _cell_content_has_table(item: NodeItem, doc: DoclingDocument) -> bool:
    """Return True if *item* is, or has a descendant that is, a TableItem."""
    if isinstance(item, TableItem):
        return True
    elif isinstance(item, NodeItem):
        for child_ref in item.children:
            if _cell_content_has_table(child_ref.resolve(doc=doc), doc):
                return True
    return False


def _mark_subtree_visited(
    item: NodeItem,
    doc: DoclingDocument,
    visited: set[str],
) -> None:
    """Recursively add *item* and all its descendants to *visited*.

    When a nested table inside a RichTableCell is flattened, its items are
    never passed through the normal serialize() path that would mark them
    visited.  Calling this keeps the visited set consistent so the document
    serializer does not emit those items again at the top level.
    """
    if isinstance(item, NodeItem):
        visited.add(item.self_ref)
        for child_ref in item.children:
            _mark_subtree_visited(child_ref.resolve(doc=doc), doc, visited)


def _collect_subtree_text(item: NodeItem, doc: DoclingDocument) -> str:
    """Collect all text from *item*'s subtree, flattening nested tables.

    Returns a space-joined string of every piece of text found so that the
    content of a nested table is preserved in a flat, readable form.

    For TableItems the text is pulled from ``data.grid`` cells directly;
    children are *not* recursed into because they duplicate the grid content
    for RichTableCells.  For all other items, ``.text`` is collected and
    children are visited recursively.
    """
    parts: list[str] = []

    if isinstance(item, TableItem):
        for row in item.data.grid:
            for cell in row:
                if cell.text:
                    parts.append(cell.text)
        return " ".join(parts)

    if isinstance(item, TextItem) and item.text:
        parts.append(item.text)

    if isinstance(item, NodeItem):
        for child_ref in item.children:
            child = child_ref.resolve(doc=doc)
            child_text = _collect_subtree_text(child, doc)
            if child_text:
                parts.append(child_text)

    return " ".join(parts)


class OrigListItemMarkerMode(str, Enum):
    """Display mode for original list item marker."""

    NEVER = "never"
    ALWAYS = "always"
    AUTO = "auto"


class FurnitureMode(str, Enum):
    """Display mode for document furniture (headers/footers)."""

    NONE = "none"
    ALL = "all"
    DISTINCT = "distinct"


class MarkdownParams(CommonParams):
    """Markdown-specific serialization parameters."""

    layers: set[ContentLayer] = {ContentLayer.BODY}
    image_mode: ImageRefMode = ImageRefMode.PLACEHOLDER
    image_placeholder: str = "<!-- image -->"
    enable_chart_tables: bool = True
    indent: int = 4
    wrap_width: Optional[PositiveInt] = None
    page_break_placeholder: Optional[str] = None  # e.g. "<!-- page break -->"
    escape_underscores: bool = True
    escape_html: bool = True
    mark_meta: bool = Field(default=False, description="Mark meta sections.")
    include_annotations: bool = Field(
        default=True,
        description="Include item annotations.",
        deprecated="Use include_meta instead.",
    )
    mark_annotations: bool = Field(
        default=False,
        description="Mark annotation sections.",
        deprecated="Use mark_meta instead.",
    )
    orig_list_item_marker_mode: OrigListItemMarkerMode = OrigListItemMarkerMode.AUTO
    ensure_valid_list_item_marker: bool = True
    format_code_blocks: bool = Field(
        default=True,
        description="Whether to wrap code items in markdown code block formatting (```). ",
    )
    compact_tables: Annotated[
        bool,
        Field(
            description=(
                "Whether to use compact table format without column padding. "
                "When False (default), tables use padded columns for better visual formatting. "
                "When True, tables use minimal whitespace, which is better for large tables and downstream processing."
            )
        ),
    ] = False
    furniture_mode: FurnitureMode = Field(
        default=FurnitureMode.NONE,
        description="Control whether headers and footers are serialized (none, all, or distinct).",
    )

    @model_validator(mode="after")
    def _adjust_layers_for_furniture(self) -> "MarkdownParams":
        """Automatically include FURNITURE layer if mode is not NONE."""
        if self.furniture_mode != FurnitureMode.NONE:
            self.layers.add(ContentLayer.FURNITURE)
        return self


class MarkdownTextSerializer(BaseModel, BaseTextSerializer):
    """Markdown-specific text item serializer."""

    @override
    def serialize(
        self,
        *,
        item: TextItem,
        doc_serializer: BaseDocSerializer,
        doc: DoclingDocument,
        is_inline_scope: bool = False,
        visited: Optional[set[str]] = None,  # refs of visited items
        **kwargs: Any,
    ) -> SerializationResult:
        """Serializes the passed item."""
        my_visited = visited if visited is not None else set()
        params = MarkdownParams(**kwargs)
        res_parts: list[SerializationResult] = []
        escape_html = True
        escape_underscores = True

        has_inline_repr = (
            item.text == ""
            and len(item.children) == 1
            and isinstance((child_group := item.children[0].resolve(doc)), InlineGroup)
        )
        if has_inline_repr:
            text = doc_serializer.serialize(item=child_group, visited=my_visited).text
            processing_pending = False
        else:
            text = item.text
            processing_pending = True

        if item.label == DocItemLabel.CHECKBOX_SELECTED:
            text = f"- [x] {text}"
        if item.label == DocItemLabel.CHECKBOX_UNSELECTED:
            text = f"- [ ] {text}"
        if isinstance(item, ListItem | TitleItem | SectionHeaderItem):
            if not has_inline_repr:
                # case where processing/formatting should be applied first (in inner scope)
                text = doc_serializer.post_process(
                    text=text,
                    escape_html=escape_html,
                    escape_underscores=escape_underscores,
                    formatting=item.formatting,
                    hyperlink=item.hyperlink,
                )
                processing_pending = False

            if isinstance(item, ListItem):
                pieces: list[str] = []
                case_auto = params.orig_list_item_marker_mode == OrigListItemMarkerMode.AUTO and bool(
                    re.search(r"[a-zA-Z0-9]", item.marker)
                )
                case_already_valid = (
                    params.ensure_valid_list_item_marker
                    and params.orig_list_item_marker_mode != OrigListItemMarkerMode.NEVER
                    and (item.marker in ["-", "*", "+"] or re.fullmatch(r"\d+\.", item.marker))
                )

                # wrap with outer marker (if applicable)
                if params.ensure_valid_list_item_marker and not case_already_valid:
                    md_marker = "-"
                    if item.parent is None:
                        _logger.warning(f"ListItem {item} must have a parent")
                    else:
                        list_group = item.parent.resolve(doc)
                        if not isinstance(list_group, ListGroup):
                            _logger.warning(f"Expected ListGroup, got {type(list_group)}")
                        elif list_group.first_item_is_enumerated(doc) and (
                            params.orig_list_item_marker_mode != OrigListItemMarkerMode.AUTO or not item.marker
                        ):
                            pos = -1
                            for i, child in enumerate(list_group.children):
                                if child.resolve(doc) == item:
                                    pos = i
                                    break
                            md_marker = f"{pos + 1}."
                    pieces.append(md_marker)

                # include original marker (if applicable)
                if item.marker and (
                    params.orig_list_item_marker_mode == OrigListItemMarkerMode.ALWAYS
                    or case_auto
                    or case_already_valid
                ):
                    pieces.append(item.marker)

                pieces.append(text)
                text_part = " ".join(pieces)
            else:
                text_part = self._format_heading(text, item)
        elif isinstance(item, CodeItem):
            if params.format_code_blocks:
                # inline items and all hyperlinks: use single backticks
                bt = is_inline_scope or (params.include_hyperlinks and item.hyperlink)
                text_part = f"`{text}`" if bt else f"```\n{text}\n```"
            else:
                text_part = text
            escape_html = False
            escape_underscores = False
        elif isinstance(item, FormulaItem):
            if text:
                text_part = f"${text}$" if is_inline_scope else f"$${text}$$"
            elif item.orig:
                text_part = "<!-- formula-not-decoded -->"
            else:
                text_part = ""
            escape_html = False
            escape_underscores = False
        elif params.wrap_width:
            # although wrapping is not guaranteed if post-processing makes changes
            text_part = textwrap.fill(text, width=params.wrap_width)
        else:
            text_part = text

        if text_part:
            text_res = create_ser_result(text=text_part, span_source=item)
            res_parts.append(text_res)

        if isinstance(item, FloatingItem):
            cap_res = doc_serializer.serialize_captions(item=item, **kwargs)
            if cap_res.text:
                res_parts.append(cap_res)

        text = (" " if is_inline_scope else "\n\n").join([r.text for r in res_parts])
        if processing_pending:
            text = doc_serializer.post_process(
                text=text,
                escape_html=escape_html,
                escape_underscores=escape_underscores,
                formatting=item.formatting,
                hyperlink=item.hyperlink,
            )
        return create_ser_result(text=text, span_source=res_parts)

    def _format_heading(
        self,
        text: str,
        item: Union[TitleItem, SectionHeaderItem],
    ) -> str:
        """Format a heading/title item. Override to customize heading representation."""
        num_hashes = 1 if isinstance(item, TitleItem) else item.level + 1
        return f"{num_hashes * '#'} {text}"


class MarkdownMetaSerializer(BaseModel, BaseMetaSerializer):
    """Markdown-specific meta serializer."""

    @override
    def serialize(
        self,
        *,
        item: NodeItem,
        doc: DoclingDocument,
        **kwargs: Any,
    ) -> SerializationResult:
        """Serialize the item's meta."""
        params = MarkdownParams(**kwargs)
        return create_ser_result(
            text="\n\n".join(
                [
                    tmp
                    for key in (list(item.meta.__class__.model_fields) + list(item.meta.get_custom_part()))
                    if (
                        (params.allowed_meta_names is None or key in params.allowed_meta_names)
                        and (key not in params.blocked_meta_names)
                        and (tmp := self._serialize_meta_field(item.meta, key, params.mark_meta))
                    )
                ]
                if item.meta
                else []
            ),
            span_source=item if isinstance(item, DocItem) else [],
            # NOTE for now using an empty span source for GroupItems
        )

    def _serialize_meta_field(self, meta: BaseMeta, name: str, mark_meta: bool) -> Optional[str]:
        if (field_val := getattr(meta, name)) is not None:
            if isinstance(field_val, SummaryMetaField):
                txt = field_val.text
            elif isinstance(field_val, KeywordsMetaField | TopicsMetaField):
                txt = ", ".join(field_val.values)
            elif isinstance(field_val, DescriptionMetaField):
                txt = field_val.text
            elif isinstance(field_val, PictureClassificationMetaField):
                txt = self._humanize_text(field_val.get_main_prediction().class_name)
            elif isinstance(field_val, MoleculeMetaField):
                txt = field_val.smi
            elif isinstance(field_val, TabularChartMetaField):
                temp_doc = DoclingDocument(name="temp")
                temp_table = temp_doc.add_table(data=field_val.chart_data)
                table_content = temp_table.export_to_markdown(temp_doc).strip()
                if table_content:
                    txt = table_content
                else:
                    return None
            elif tmp := str(field_val or ""):
                txt = tmp
            else:
                return None
            return f"[{self._humanize_text(name, title=True)}] {txt}" if mark_meta else txt
        else:
            return None


class MarkdownAnnotationSerializer(BaseModel, BaseAnnotationSerializer):
    """Markdown-specific annotation serializer."""

    @override
    def serialize(
        self,
        *,
        item: DocItem,
        doc: DoclingDocument,
        **kwargs: Any,
    ) -> SerializationResult:
        """Serialize the item's annotations."""
        params = MarkdownParams(**kwargs)

        res_parts: list[SerializationResult] = []
        for ann in item.get_annotations():
            if isinstance(
                ann,
                PictureClassificationData | DescriptionAnnotation | PictureMoleculeData,
            ):
                if ann_text := _get_annotation_text(ann):
                    ann_res = create_ser_result(
                        text=(
                            (f'<!--<annotation kind="{ann.kind}">-->{ann_text}<!--<annotation/>-->')
                            if params.mark_annotations
                            else ann_text
                        ),
                        span_source=item,
                    )
                    res_parts.append(ann_res)
        return create_ser_result(
            text="\n\n".join([r.text for r in res_parts if r.text]),
            span_source=item,
        )


class MarkdownTableSerializer(BaseTableSerializer):
    """Markdown-specific table item serializer."""

    @override
    def get_header_and_body_lines(
        self,
        *,
        table_text: str,
        **kwargs: Any,
    ) -> tuple[list[str], list[str]]:
        """Get header lines and body lines from the markdown table.

        Returns:
            A tuple of (header_lines, body_lines) where header_lines contains
            the header row and separator row, and body_lines contains the data rows.
        """
        lines = [line for line in table_text.splitlines(True) if line.strip()]

        if len(lines) < 2:
            # Not enough lines for a proper markdown table (need at least header + separator)
            return [], lines

        # In markdown tables:
        # Line 0: Header row
        # Line 1: Separator row (with dashes)
        # Lines 2+: Body rows
        header_lines = lines[:2]
        body_lines = lines[2:]

        return header_lines, body_lines

    @staticmethod
    def _compact_table(table_text: str) -> str:
        """Remove padding from a markdown table.

        Args:
            table_text: Padded markdown table string

        Returns:
            Compact markdown table string
        """
        lines = table_text.split("\n")
        compact_lines = []

        for i, line in enumerate(lines):
            if not line:
                continue

            parts = line.split("|")[1:-1]

            # For separator line (second line), preserve alignment marks
            if i == 1:
                compact_parts = []
                for part in parts:
                    p = part.strip()
                    if p.startswith(":") and p.endswith(":"):
                        compact_parts.append(":-:")
                    elif p.startswith(":"):
                        compact_parts.append(":-")
                    elif p.endswith(":"):
                        compact_parts.append("-:")
                    else:
                        compact_parts.append("-")
            else:
                compact_parts = [part.strip() for part in parts]

            compact_lines.append("| " + " | ".join(compact_parts) + " |")

        return "\n".join(compact_lines)

    @override
    def serialize(
        self,
        *,
        item: TableItem,
        doc_serializer: BaseDocSerializer,
        doc: DoclingDocument,
        **kwargs: Any,
    ) -> SerializationResult:
        """Serializes the passed item."""
        if kwargs.get("_nested_in_table"):
            visited: set[str] = kwargs.get("visited") or set()
            _mark_subtree_visited(item, doc, visited)
            return create_ser_result(
                text=_collect_subtree_text(item, doc),
                span_source=item,
            )

        params = MarkdownParams(**kwargs)
        res_parts: list[SerializationResult] = []

        cap_res = doc_serializer.serialize_captions(
            item=item,
            **kwargs,
        )
        if cap_res.text:
            res_parts.append(cap_res)

        if item.self_ref not in doc_serializer.get_excluded_refs(**kwargs):
            if _should_use_legacy_annotations(params=params, item=item):
                ann_res = doc_serializer.serialize_annotations(
                    item=item,
                    **kwargs,
                )
                if ann_res.text:
                    res_parts.append(ann_res)

            rows = []
            for row in item.data.grid:
                rendered_row = []
                for col in row:
                    if isinstance(col, RichTableCell):
                        ref_item = col.ref.resolve(doc=doc)
                        inner_kwargs = {**kwargs, "_nested_in_table": True}
                        cell_text = doc_serializer.serialize(
                            item=ref_item,
                            **inner_kwargs,
                        ).text
                    else:
                        cell_text = col.text or ""
                    # Newlines and pipes must be escaped in every cell so the
                    # markdown table stays valid.
                    rendered_row.append(cell_text.replace("\n", " ").replace("|", "&#124;"))
                rows.append(rendered_row)
            if len(rows) > 0:
                # Always disable numparse to prevent silent precision loss in numeric values
                # Use tabulate's _column_type to detect numeric columns for right-alignment
                colalign = []
                if len(rows) > 1:  # Need at least header + 1 data row
                    num_cols = len(rows[0])
                    for col_idx in range(num_cols):
                        col_values = [row[col_idx] if col_idx < len(row) else "" for row in rows[1:]]
                        col_type = _column_type(col_values)
                        colalign.append("right" if col_type in (int, float) else "left")
                table_text = tabulate(
                    rows[1:],
                    headers=rows[0],
                    tablefmt="github",
                    disable_numparse=True,
                    colalign=tuple(colalign) if colalign else None,
                )

                if params.compact_tables:
                    table_text = self._compact_table(table_text)
            else:
                table_text = ""
            if table_text:
                res_parts.append(create_ser_result(text=table_text, span_source=item))

        text_res = "\n\n".join([r.text for r in res_parts])

        return create_ser_result(text=text_res, span_source=res_parts)


class MarkdownPictureSerializer(BasePictureSerializer):
    """Markdown-specific picture item serializer."""

    @override
    def serialize(
        self,
        *,
        item: PictureItem,
        doc_serializer: BaseDocSerializer,
        doc: DoclingDocument,
        **kwargs: Any,
    ) -> SerializationResult:
        """Serializes the passed item."""
        params = MarkdownParams(**kwargs)

        res_parts: list[SerializationResult] = []

        cap_res = doc_serializer.serialize_captions(
            item=item,
            **kwargs,
        )
        if cap_res.text:
            res_parts.append(cap_res)

        if item.self_ref not in doc_serializer.get_excluded_refs(**kwargs):
            if _should_use_legacy_annotations(params=params, item=item):
                ann_res = doc_serializer.serialize_annotations(
                    item=item,
                    **kwargs,
                )
                if ann_res.text:
                    res_parts.append(ann_res)

            img_res = self._serialize_image_part(
                item=item,
                doc=doc,
                image_mode=params.image_mode,
                image_placeholder=params.image_placeholder,
            )
            if img_res.text:
                res_parts.append(img_res)

        if params.enable_chart_tables and _should_use_legacy_annotations(
            params=params,
            item=item,
            kind=PictureTabularChartData.model_fields["kind"].default,
        ):
            # Check if picture has attached PictureTabularChartData
            tabular_chart_annotations = [ann for ann in item.annotations if isinstance(ann, PictureTabularChartData)]
            if len(tabular_chart_annotations) > 0:
                temp_doc = DoclingDocument(name="temp")
                temp_table = temp_doc.add_table(data=tabular_chart_annotations[0].chart_data)
                md_table_content = temp_table.export_to_markdown(temp_doc)
                if len(md_table_content) > 0:
                    res_parts.append(create_ser_result(text=md_table_content, span_source=item))
        text_res = "\n\n".join([r.text for r in res_parts if r.text])

        return create_ser_result(text=text_res, span_source=res_parts)

    def _serialize_image_part(
        self,
        item: PictureItem,
        doc: DoclingDocument,
        image_mode: ImageRefMode,
        image_placeholder: str,
        **kwargs: Any,
    ) -> SerializationResult:
        error_response = (
            "<!-- 🖼️❌ Image not available. Please use `PdfPipelineOptions(generate_picture_images=True)` -->"
        )
        if image_mode == ImageRefMode.PLACEHOLDER:
            text_res = image_placeholder
        elif image_mode == ImageRefMode.EMBEDDED:
            # short-cut: we already have the image in base64
            if (
                isinstance(item.image, ImageRef)
                and isinstance(item.image.uri, AnyUrl)
                and item.image.uri.scheme == "data"
            ):
                text = f"![Image]({item.image.uri})"
                text_res = text
            else:
                # get the item.image._pil or crop it out of the page-image
                img = item.get_image(doc=doc)

                if img is not None:
                    imgb64 = item._image_to_base64(img)
                    text = f"![Image](data:image/png;base64,{imgb64})"

                    text_res = text
                else:
                    text_res = error_response
        elif image_mode == ImageRefMode.REFERENCED:
            if not isinstance(item.image, ImageRef) or (
                isinstance(item.image.uri, AnyUrl) and item.image.uri.scheme == "data"
            ):
                text_res = image_placeholder
            else:
                text_res = f"![Image]({item.image.uri!s})"
        else:
            text_res = image_placeholder

        return create_ser_result(text=text_res, span_source=item)


class MarkdownKeyValueSerializer(BaseKeyValueSerializer):
    """Markdown-specific key-value item serializer."""

    @override
    def serialize(
        self,
        *,
        item: KeyValueItem,
        doc_serializer: "BaseDocSerializer",
        doc: DoclingDocument,
        **kwargs: Any,
    ) -> SerializationResult:
        """Serializes the passed item."""
        # TODO add actual implementation
        if item.self_ref not in doc_serializer.get_excluded_refs():
            return create_ser_result(
                text="<!-- missing-key-value-item -->",
                span_source=item,
            )
        else:
            return create_ser_result()


class MarkdownFormSerializer(BaseFormSerializer):
    """Markdown-specific form item serializer."""

    @override
    def serialize(
        self,
        *,
        item: FormItem,
        doc_serializer: "BaseDocSerializer",
        doc: DoclingDocument,
        **kwargs: Any,
    ) -> SerializationResult:
        """Serializes the passed item."""
        # TODO add actual implementation
        if item.self_ref not in doc_serializer.get_excluded_refs():
            return create_ser_result(
                text="<!-- missing-form-item -->",
                span_source=item,
            )
        else:
            return create_ser_result()


class MarkdownListSerializer(BaseModel, BaseListSerializer):
    """Markdown-specific list serializer."""

    @override
    def serialize(
        self,
        *,
        item: ListGroup,
        doc_serializer: "BaseDocSerializer",
        doc: DoclingDocument,
        list_level: int = 0,
        is_inline_scope: bool = False,
        visited: Optional[set[str]] = None,  # refs of visited items
        **kwargs: Any,
    ) -> SerializationResult:
        """Serializes the passed item."""
        params = MarkdownParams(**kwargs)
        my_visited = visited if visited is not None else set()
        parts = doc_serializer.get_parts(
            item=item,
            list_level=list_level + 1,
            is_inline_scope=is_inline_scope,
            visited=my_visited,
            **kwargs,
        )
        sep = "\n"
        my_parts: list[SerializationResult] = []
        for p in parts:
            if (
                my_parts
                and p.text
                and p.spans
                and p.spans[0].item.parent
                and isinstance(p.spans[0].item.parent.resolve(doc), InlineGroup)
            ):
                my_parts[-1].text = f"{my_parts[-1].text}{p.text}"  # append to last
                my_parts[-1].spans.extend(p.spans)
            else:
                my_parts.append(p)

        indent_str = list_level * params.indent * " "
        text_res = sep.join(
            [
                # avoid additional marker on already evaled sublists
                (c.text if c.text and c.text[0] == " " else f"{indent_str}{c.text}")
                for c in my_parts
            ]
        )
        return create_ser_result(text=text_res, span_source=my_parts)


class MarkdownInlineSerializer(BaseInlineSerializer):
    """Markdown-specific inline group serializer."""

    @override
    def serialize(
        self,
        *,
        item: InlineGroup,
        doc_serializer: "BaseDocSerializer",
        doc: DoclingDocument,
        list_level: int = 0,
        visited: Optional[set[str]] = None,  # refs of visited items
        **kwargs: Any,
    ) -> SerializationResult:
        """Serializes the passed item."""
        my_visited = visited if visited is not None else set()
        parts = doc_serializer.get_parts(
            item=item,
            list_level=list_level,
            is_inline_scope=True,
            visited=my_visited,
            **kwargs,
        )
        text_res = " ".join([p.text for p in parts if p.text])
        return create_ser_result(text=text_res, span_source=parts)


class MarkdownFallbackSerializer(BaseFallbackSerializer):
    """Markdown-specific fallback serializer."""

    @override
    def serialize(
        self,
        *,
        item: NodeItem,
        doc_serializer: "BaseDocSerializer",
        doc: DoclingDocument,
        **kwargs: Any,
    ) -> SerializationResult:
        """Serializes the passed item."""
        if isinstance(item, GroupItem):
            parts = doc_serializer.get_parts(item=item, **kwargs)
            text_res = "\n\n".join([p.text for p in parts if p.text])
            return create_ser_result(text=text_res, span_source=parts)
        else:
            return create_ser_result(
                text="<!-- missing-text -->",
                span_source=item if isinstance(item, DocItem) else [],
            )


class MarkdownDocSerializer(DocSerializer):
    """Markdown-specific document serializer."""

    text_serializer: BaseTextSerializer = MarkdownTextSerializer()
    table_serializer: BaseTableSerializer = MarkdownTableSerializer()
    picture_serializer: BasePictureSerializer = MarkdownPictureSerializer()
    key_value_serializer: BaseKeyValueSerializer = MarkdownKeyValueSerializer()
    form_serializer: BaseFormSerializer = MarkdownFormSerializer()
    fallback_serializer: BaseFallbackSerializer = MarkdownFallbackSerializer()

    list_serializer: BaseListSerializer = MarkdownListSerializer()
    inline_serializer: BaseInlineSerializer = MarkdownInlineSerializer()

    meta_serializer: BaseMetaSerializer = MarkdownMetaSerializer()
    annotation_serializer: BaseAnnotationSerializer = MarkdownAnnotationSerializer()

    params: MarkdownParams = MarkdownParams()

    @override
    def serialize_bold(self, text: str, **kwargs: Any):
        """Apply Markdown-specific bold serialization."""
        return f"**{text}**"

    @override
    def serialize_italic(self, text: str, **kwargs: Any):
        """Apply Markdown-specific italic serialization."""
        return f"*{text}*"

    @override
    def serialize_strikethrough(self, text: str, **kwargs: Any):
        """Apply Markdown-specific strikethrough serialization."""
        return f"~~{text}~~"

    @override
    def serialize_hyperlink(
        self,
        text: str,
        hyperlink: Union[AnyUrl, Path],
        **kwargs: Any,
    ):
        """Apply Markdown-specific hyperlink serialization."""
        return f"[{text}]({hyperlink!s})"

    @classmethod
    def _escape_underscores(cls, text: str):
        """Escape underscores but leave them intact in the URL.."""
        # Firstly, identify all the URL patterns.
        url_pattern = r"!\[.*?\]\((.*?)\)"

        parts = []
        last_end = 0

        for match in re.finditer(url_pattern, text):
            # Text to add before the URL (needs to be escaped)
            before_url = text[last_end : match.start()]
            parts.append(re.sub(r"(?<!\\)_", r"\_", before_url))

            # Add the full URL part (do not escape)
            parts.append(match.group(0))
            last_end = match.end()

        # Add the final part of the text (which needs to be escaped)
        if last_end < len(text):
            parts.append(re.sub(r"(?<!\\)_", r"\_", text[last_end:]))

        return "".join(parts)
        # return text.replace("_", r"\_")

    def post_process(
        self,
        text: str,
        *,
        escape_html: bool = True,
        escape_underscores: bool = True,
        formatting: Optional[Formatting] = None,
        hyperlink: Optional[Union[AnyUrl, Path]] = None,
        **kwargs: Any,
    ) -> str:
        """Apply some text post-processing steps."""
        res = text
        params = self.params.merge_with_patch(patch=kwargs)
        if escape_underscores and params.escape_underscores:
            res = self._escape_underscores(text)
        if escape_html and params.escape_html:
            res = html.escape(res, quote=False)
        res = super().post_process(
            text=res,
            formatting=formatting,
            hyperlink=hyperlink,
        )
        return res

    @override
    def serialize_doc(
        self,
        *,
        parts: list[SerializationResult],
        **kwargs: Any,
    ) -> SerializationResult:
        """Serialize a document out of its parts."""
        text_res = "\n\n".join([p.text for p in parts if p.text])
        if self.requires_page_break():
            page_sep = self.params.page_break_placeholder or ""
            for full_match, _, _ in self._get_page_breaks(text=text_res):
                text_res = text_res.replace(full_match, page_sep)

        return create_ser_result(text=text_res, span_source=parts)

    @override
    def requires_page_break(self) -> bool:
        """Whether to add page breaks."""
        return self.params.page_break_placeholder is not None

    @override
    def serialize(
        self,
        *,
        item: Optional[NodeItem] = None,
        list_level: int = 0,
        is_inline_scope: bool = False,
        visited: Optional[set[str]] = None,
        **kwargs: Any,
    ) -> SerializationResult:
        """Serialize a given node."""

        if item is not None and getattr(item, "content_layer", None) == ContentLayer.FURNITURE:
            # 'none' mode: skip completely
            if self.params.furniture_mode == FurnitureMode.NONE:
                return create_ser_result()

            # 'distinct' mode: deduplicate based on text
            if self.params.furniture_mode == FurnitureMode.DISTINCT:
                # Initialize the tracking set on the serializer instance if it doesn't exist
                if not hasattr(self, "_seen_furniture"):
                    self._seen_furniture: set[str] = set()

                # Extract text for comparison (fallback to ref if no text exists)
                item_text = getattr(item, "text", None)
                if not item_text:
                    item_text = item.self_ref

                # If we've seen this exact header/footer text before, skip it
                if item_text in self._seen_furniture:
                    return create_ser_result()

                # Otherwise, mark it as seen and proceed
                self._seen_furniture.add(item_text)

        return super().serialize(
            item=item,
            list_level=list_level,
            is_inline_scope=is_inline_scope,
            visited=visited,
            **(dict(delim="\n\n") | kwargs),
        )