elements.py

from __future__ import annotations

import abc
import copy
import dataclasses as dc
import enum
import functools
import hashlib
import os
import pathlib
import uuid
from functools import cached_property
from itertools import groupby
from types import MappingProxyType
from typing import Any, Callable, FrozenSet, Optional, Sequence, cast

from typing_extensions import ParamSpec, TypeAlias, TypedDict

from unstructured.documents.coordinates import (
    TYPE_TO_COORDINATE_SYSTEM_MAP,
    CoordinateSystem,
    RelativeCoordinateSystem,
)
from unstructured.partition.utils.constants import UNSTRUCTURED_INCLUDE_DEBUG_METADATA
from unstructured.utils import get_call_args_applying_defaults

Point: TypeAlias = "tuple[float, float]"
Points: TypeAlias = "tuple[Point, ...]"


@dc.dataclass
class DataSourceMetadata:
    """Metadata fields that pertain to the data source of the document."""

    url: Optional[str] = None
    version: Optional[str] = None
    record_locator: Optional[dict[str, Any]] = None  # Values must be JSON-serializable
    date_created: Optional[str] = None
    date_modified: Optional[str] = None
    date_processed: Optional[str] = None
    permissions_data: Optional[list[dict[str, Any]]] = None

    def to_dict(self):
        return {key: value for key, value in self.__dict__.items() if value is not None}

    @classmethod
    def from_dict(cls, input_dict: dict[str, Any]):
        # Only use existing fields when constructing
        supported_fields = [f.name for f in dc.fields(cls)]
        args = {k: v for k, v in input_dict.items() if k in supported_fields}

        return cls(**args)


@dc.dataclass
class CoordinatesMetadata:
    """Metadata fields that pertain to the coordinates of the element."""

    points: Optional[Points]
    system: Optional[CoordinateSystem]

    def __init__(self, points: Optional[Points], system: Optional[CoordinateSystem]):
        # Both `points` and `system` must be present; one is not meaningful without the other.
        if (points is None and system is not None) or (points is not None and system is None):
            raise ValueError(
                "Coordinates points should not exist without coordinates system and vice versa.",
            )
        self.points = points
        self.system = system

    def __eq__(self, other: Any) -> bool:
        if not isinstance(other, CoordinatesMetadata):
            return False
        return all(
            [
                (self.points == other.points),
                (self.system == other.system),
            ],
        )

    def to_dict(self):
        return {
            "points": self.points,
            "system": None if self.system is None else str(self.system.__class__.__name__),
            "layout_width": None if self.system is None else self.system.width,
            "layout_height": None if self.system is None else self.system.height,
        }

    @classmethod
    def from_dict(cls, input_dict: dict[str, Any]):
        # `input_dict` may contain a tuple of tuples or a list of lists
        def convert_to_points(sequence_of_sequences: Sequence[Sequence[float]]) -> Points:
            points: list[Point] = []
            for seq in sequence_of_sequences:
                if isinstance(seq, list):
                    points.append(cast(Point, tuple(seq)))
                elif isinstance(seq, tuple):
                    points.append(cast(Point, seq))
            return tuple(points)

        # -- parse points --
        input_points = input_dict.get("points")
        points = convert_to_points(input_points) if input_points is not None else None

        # -- parse system --
        system_name = input_dict.get("system")
        width = input_dict.get("layout_width")
        height = input_dict.get("layout_height")
        system = (
            None
            if system_name is None
            else (
                RelativeCoordinateSystem()
                if system_name == "RelativeCoordinateSystem"
                else (
                    TYPE_TO_COORDINATE_SYSTEM_MAP[system_name](width, height)
                    if (
                        width is not None
                        and height is not None
                        and system_name in TYPE_TO_COORDINATE_SYSTEM_MAP
                    )
                    else None
                )
            )
        )

        return cls(points=points, system=system)


class Link(TypedDict):
    """Metadata related to extracted links"""

    text: Optional[str]
    url: str
    start_index: int


class FormKeyOrValue(TypedDict):
    text: str
    layout_element_id: Optional[str]
    custom_element: Optional[Text]


class FormKeyValuePair(TypedDict):
    key: FormKeyOrValue
    value: Optional[FormKeyOrValue]
    confidence: float


class ElementMetadata:
    """Fully-dynamic replacement for dataclass-based ElementMetadata."""

    # NOTE(scanny): To add a field:
    # - Add the field declaration with type here at the top. This makes it a "known" field and
    #   enables type-checking and completion.
    # - Add a parameter with default for field in __init__() and assign it in __init__() body.
    # - Add a consolidation strategy for the new field in
    #   `ConsolidationStrategy.field_consolidation_strategies()` below. This strategy will be used
    #   to consolidate this new metadata field from each pre-chunk element during chunking.
    # - Add field-name to DEBUG_FIELD_NAMES if it shouldn't appear in dict/JSON or participate in
    #   equality comparison.

    attached_to_filename: Optional[str]
    category_depth: Optional[int]
    coordinates: Optional[CoordinatesMetadata]
    data_source: Optional[DataSourceMetadata]
    # -- True when PDF content-stream text includes characters rendered invisibly. --
    contains_invisible_text: Optional[bool]
    # -- Detection Model Class Probabilities from Unstructured-Inference Hi-Res --
    detection_class_prob: Optional[float]
    # -- DEBUG field, the detection mechanism that emitted this element --
    detection_origin: Optional[str]
    # -- per-attribute model provenance for enrichments. Maps a written attribute name (e.g.
    # -- "text", "text_as_html", "embeddings") to a list of records in application order, each
    # -- {"type", "provider", "model"}. Authoring enrichments overwrite (reset the list);
    # -- additive enrichments append (preserving the prior author). --
    enrichment_origins: Optional[dict[str, list[dict[str, str]]]]
    emphasized_text_contents: Optional[list[str]]
    emphasized_text_tags: Optional[list[str]]
    file_directory: Optional[str]
    filename: Optional[str]
    filetype: Optional[str]
    image_url: Optional[str]
    image_path: Optional[str]
    image_base64: Optional[str]
    image_mime_type: Optional[str]
    # -- specific to DOCX which has distinct primary, first-page, and even-page header/footers --
    header_footer_type: Optional[str]
    # -- used in chunks only, when chunk must be split mid-text to fit window --
    is_continuation: Optional[bool]
    key_value_pairs: Optional[list[FormKeyValuePair]]
    languages: Optional[list[str]]
    last_modified: Optional[str]
    link_texts: Optional[list[str]]
    link_urls: Optional[list[str]]
    link_start_indexes: Optional[list[int]]
    links: Optional[list[Link]]
    # -- used in chunks only, allowing access to element(s) chunk was formed from when enabled --
    orig_elements: Optional[list[Element]]
    # -- the worksheet name in XLXS documents --
    page_name: Optional[str]
    # -- page numbers currently supported for DOCX, HTML, PDF, and PPTX documents --
    page_number: Optional[int]
    parent_id: Optional[str]

    # -- routing decision (page-level) --
    routing: Optional[str]
    routing_score: Optional[float]

    # -- e-mail specific metadata fields --
    bcc_recipient: Optional[list[str]]
    cc_recipient: Optional[list[str]]
    email_message_id: Optional[str]
    sent_from: Optional[list[str]]
    sent_to: Optional[list[str]]
    subject: Optional[str]
    signature: Optional[str]

    # -- used for Table elements to capture rows/col structure --
    text_as_html: Optional[str]
    is_extracted: Optional[str]
    table_as_cells: Optional[dict[str, str | int]]
    table_extraction_method: Optional[str]  # "grid", "tatr", or "vlm"

    # -- used for TableChunk elements to enable table reconstruction --
    table_id: Optional[str]
    chunk_index: Optional[int]
    num_carried_over_header_rows: Optional[int]
    url: Optional[str]

    # -- speech-to-text segment timestamps (seconds) when element is from partition_audio --
    segment_end_seconds: Optional[float]
    segment_start_seconds: Optional[float]

    # -- debug fields can be assigned and referenced using dotted-notation but are not serialized
    # -- to dict/JSON, do not participate in equality comparison, and are not included in the
    # -- `.fields` dict used by other parts of the library like chunking and weaviate.
    DEBUG_FIELD_NAMES = frozenset(["detection_origin"])

    def __init__(
        self,
        attached_to_filename: Optional[str] = None,
        bcc_recipient: Optional[list[str]] = None,
        category_depth: Optional[int] = None,
        cc_recipient: Optional[list[str]] = None,
        contains_invisible_text: Optional[bool] = None,
        coordinates: Optional[CoordinatesMetadata] = None,
        data_source: Optional[DataSourceMetadata] = None,
        detection_class_prob: Optional[float] = None,
        enrichment_origins: Optional[dict[str, list[dict[str, str]]]] = None,
        emphasized_text_contents: Optional[list[str]] = None,
        emphasized_text_tags: Optional[list[str]] = None,
        file_directory: Optional[str] = None,
        filename: Optional[str | pathlib.Path] = None,
        filetype: Optional[str] = None,
        header_footer_type: Optional[str] = None,
        image_base64: Optional[str] = None,
        image_mime_type: Optional[str] = None,
        image_url: Optional[str] = None,
        image_path: Optional[str] = None,
        is_continuation: Optional[bool] = None,
        languages: Optional[list[str]] = None,
        last_modified: Optional[str] = None,
        link_start_indexes: Optional[list[int]] = None,
        link_texts: Optional[list[str]] = None,
        link_urls: Optional[list[str]] = None,
        links: Optional[list[Link]] = None,
        email_message_id: Optional[str] = None,
        orig_elements: Optional[list[Element]] = None,
        page_name: Optional[str] = None,
        page_number: Optional[int] = None,
        parent_id: Optional[str] = None,
        routing: Optional[str] = None,
        routing_score: Optional[float] = None,
        sent_from: Optional[list[str]] = None,
        sent_to: Optional[list[str]] = None,
        signature: Optional[str] = None,
        subject: Optional[str] = None,
        table_as_cells: Optional[dict[str, str | int]] = None,
        table_extraction_method: Optional[str] = None,
        table_id: Optional[str] = None,
        chunk_index: Optional[int] = None,
        num_carried_over_header_rows: Optional[int] = None,
        text_as_html: Optional[str] = None,
        url: Optional[str] = None,
        segment_end_seconds: Optional[float] = None,
        segment_start_seconds: Optional[float] = None,
    ) -> None:
        self.attached_to_filename = attached_to_filename
        self.bcc_recipient = bcc_recipient
        self.category_depth = category_depth
        self.cc_recipient = cc_recipient
        self.contains_invisible_text = contains_invisible_text
        self.coordinates = coordinates
        self.data_source = data_source
        self.detection_class_prob = detection_class_prob
        self.enrichment_origins = enrichment_origins
        self.emphasized_text_contents = emphasized_text_contents
        self.emphasized_text_tags = emphasized_text_tags

        # -- accommodate pathlib.Path for filename --
        filename = str(filename) if isinstance(filename, pathlib.Path) else filename
        # -- produces "", "" when filename arg is None --
        directory_path, file_name = os.path.split(filename or "")
        # -- prefer `file_directory` arg if specified, otherwise split of file-path passed as
        # -- `filename` arg, or None if `filename` is the empty string.
        self.file_directory = file_directory or directory_path or None
        self.filename = file_name or None

        self.filetype = filetype
        self.header_footer_type = header_footer_type
        self.image_base64 = image_base64
        self.image_mime_type = image_mime_type
        self.image_url = image_url
        self.image_path = image_path
        self.is_continuation = is_continuation
        self.languages = languages
        self.last_modified = last_modified
        self.link_texts = link_texts
        self.link_urls = link_urls
        self.link_start_indexes = link_start_indexes
        self.links = links
        self.email_message_id = email_message_id
        self.orig_elements = orig_elements
        self.page_name = page_name
        self.page_number = page_number
        self.parent_id = parent_id
        self.routing = routing
        self.routing_score = routing_score
        self.sent_from = sent_from
        self.sent_to = sent_to
        self.signature = signature
        self.subject = subject
        self.text_as_html = text_as_html
        self.table_as_cells = table_as_cells
        self.table_extraction_method = table_extraction_method
        self.table_id = table_id
        self.chunk_index = chunk_index
        self.num_carried_over_header_rows = num_carried_over_header_rows
        self.url = url
        self.segment_end_seconds = segment_end_seconds
        self.segment_start_seconds = segment_start_seconds

    def __eq__(self, other: object) -> bool:
        """Implments equivalence, like meta == other_meta.

        All fields at all levels must match. Unpopulated fields are not considered except when
        populated in one and not the other.
        """
        if not isinstance(other, ElementMetadata):
            return False
        return self.fields == other.fields

    def __getattr__(self, attr_name: str) -> None:
        """Only called when attribute doesn't exist."""
        if attr_name in self._known_field_names:
            return None
        raise AttributeError(f"'ElementMetadata' object has no attribute '{attr_name}'")

    def __setattr__(self, __name: str, __value: Any) -> None:
        if __value is None:
            # -- can't use `hasattr()` for this because it calls `__getattr__()` to find out --
            if __name in self.__dict__:
                delattr(self, __name)
            return
        if not UNSTRUCTURED_INCLUDE_DEBUG_METADATA and __name in self.DEBUG_FIELD_NAMES:
            return
        super().__setattr__(__name, __value)

    @classmethod
    def from_dict(cls, meta_dict: dict[str, Any]) -> ElementMetadata:
        """Construct from a metadata-dict.

        This would generally be a dict formed using the `.to_dict()` method and stored as JSON
        before "rehydrating" it using this method.
        """
        from unstructured.staging.base import elements_from_base64_gzipped_json

        # -- avoid unexpected mutation by working on a copy of provided dict --
        meta_dict = copy.deepcopy(meta_dict)
        self = ElementMetadata()
        for field_name, field_value in meta_dict.items():
            if field_name == "coordinates":
                self.coordinates = CoordinatesMetadata.from_dict(field_value)
            elif field_name == "data_source":
                self.data_source = DataSourceMetadata.from_dict(field_value)
            elif field_name == "orig_elements":
                self.orig_elements = elements_from_base64_gzipped_json(field_value)
            elif field_name == "key_value_pairs":
                self.key_value_pairs = _kvform_rehydrate_internal_elements(field_value)
            else:
                setattr(self, field_name, field_value)

        return self

    @property
    def fields(self) -> MappingProxyType[str, Any]:
        """Populated metadata fields in this object as a read-only dict.

        Basically `self.__dict__` but it needs a little filtering to remove entries like
        "_known_field_names". Note this is a *snapshot* and will not reflect later changes.
        """
        return MappingProxyType(
            {
                field_name: field_value
                for field_name, field_value in self.__dict__.items()
                if not field_name.startswith("_") and field_name not in self.DEBUG_FIELD_NAMES
            }
        )

    @property
    def known_fields(self) -> MappingProxyType[str, Any]:
        """Populated non-ad-hoc fields in this object as a read-only dict.

        Only fields declared at the top of this class are included. Ad-hoc fields added to this
        instance by assignment are not. Note this is a *snapshot* and will not reflect changes that
        occur after this call.
        """
        known_field_names = self._known_field_names
        return MappingProxyType(
            {
                field_name: field_value
                for field_name, field_value in self.__dict__.items()
                if (field_name in known_field_names and field_name not in self.DEBUG_FIELD_NAMES)
            }
        )

    def to_dict(self) -> dict[str, Any]:
        """Convert this metadata to dict form, suitable for JSON serialization.

        The returned dict is "sparse" in that no key-value pair appears for a field with value
        `None`.
        """
        from unstructured.staging.base import elements_to_base64_gzipped_json

        meta_dict = copy.deepcopy(dict(self.fields))

        # -- remove fields that should not be serialized --
        for field_name in self.DEBUG_FIELD_NAMES:
            meta_dict.pop(field_name, None)

        # -- don't serialize empty lists --
        meta_dict: dict[str, Any] = {
            field_name: value
            for field_name, value in meta_dict.items()
            if value != [] and value != {}
        }

        # -- serialize sub-object types when present --
        if self.coordinates is not None:
            meta_dict["coordinates"] = self.coordinates.to_dict()
        if self.data_source is not None:
            meta_dict["data_source"] = self.data_source.to_dict()
        if self.orig_elements is not None:
            meta_dict["orig_elements"] = elements_to_base64_gzipped_json(self.orig_elements)
        if self.key_value_pairs is not None:
            meta_dict["key_value_pairs"] = _kvform_pairs_to_dict(self.key_value_pairs)

        return meta_dict

    def update(self, other: ElementMetadata) -> None:
        """Update self with all fields present in `other`.

        Semantics are like those of `dict.update()`.

        - fields present in both `self` and `other` will be updated to the value in `other`.
        - fields present in `other` but not `self` will be added to `self`.
        - fields present in `self` but not `other` are unchanged.
        - `other` is unchanged.
        - both ad-hoc and known fields participate in update with the same semantics.

        Note that fields listed in DEBUG_FIELD_NAMES are skipped in this process. Those can only be
        updated by direct assignment to the instance.
        """
        if not isinstance(other, ElementMetadata):  # pyright: ignore[reportUnnecessaryIsInstance]
            raise ValueError("argument to '.update()' must be an instance of 'ElementMetadata'")

        for field_name, field_value in other.fields.items():
            setattr(self, field_name, field_value)

    @cached_property
    def _known_field_names(self) -> FrozenSet[str]:
        """field-names for non-user-defined fields, available on all ElementMetadata instances.

        Note that the first call to this lazyproperty adds a `"_known_field_names"` item to the
        `__dict__` of this instance, so this be called *before* iterating through `self.__dict__`
        to avoid a mid-iteration mutation.
        """
        # -- self.__annotations__ is a dict and iterating it produces its keys, which are the
        # -- field-names we want here.
        return frozenset(self.__annotations__)


class ConsolidationStrategy(enum.Enum):
    """Methods by which a metadata field can be consolidated across a collection of elements.

    These are assigned to `ElementMetadata` field-names immediately below. Metadata consolidation is
    part of the chunking process and may arise elsewhere as well.
    """

    DROP = "drop"
    """Do not include this field in the consolidated metadata object."""

    FIRST = "first"
    """Use the first value encountered, omit if not present in any elements."""

    STRING_CONCATENATE = "string_concatenate"
    """Combine the values of this field across elements. Only suitable for fields of `str` type."""

    LIST_CONCATENATE = "LIST_CONCATENATE"
    """Concatenate the list values across elements. Only suitable for fields of `List` type."""

    LIST_UNIQUE = "list_unique"
    """Union list values across elements, preserving order. Only suitable for `List` fields."""

    DICT_LIST_UNIQUE = "dict_list_unique"
    """Merge dict-of-list values across elements: union keys, and per key concatenate the lists
    then drop duplicate records, preserving first-seen order. Suitable for `dict[str, list]`
    fields like `enrichment_origins`."""

    ANY = "any"
    """Use True when any consolidated field value is truthy."""

    @classmethod
    def field_consolidation_strategies(cls) -> dict[str, ConsolidationStrategy]:
        """Mapping from ElementMetadata field-name to its consolidation strategy.

        Note that only _TextSection objects ("pre-chunks" containing only `Text` elements that are
        not `Table`) have their metadata consolidated, so these strategies are only applicable for
        non-Table Text elements.
        """
        return {
            "attached_to_filename": cls.FIRST,
            "cc_recipient": cls.FIRST,
            "bcc_recipient": cls.FIRST,
            "category_depth": cls.DROP,
            "contains_invisible_text": cls.ANY,
            "coordinates": cls.DROP,
            "data_source": cls.FIRST,
            "detection_class_prob": cls.DROP,
            "detection_origin": cls.DROP,
            "enrichment_origins": cls.DICT_LIST_UNIQUE,
            "emphasized_text_contents": cls.LIST_CONCATENATE,
            "emphasized_text_tags": cls.LIST_CONCATENATE,
            "file_directory": cls.FIRST,
            "filename": cls.FIRST,
            "filetype": cls.FIRST,
            "header_footer_type": cls.DROP,
            "image_url": cls.DROP,
            "image_path": cls.DROP,
            "image_base64": cls.DROP,
            "image_mime_type": cls.DROP,
            "is_continuation": cls.DROP,  # -- not expected, added by chunking, not before --
            "is_extracted": cls.DROP,
            "languages": cls.LIST_UNIQUE,
            "last_modified": cls.FIRST,
            "link_texts": cls.LIST_CONCATENATE,
            "link_urls": cls.LIST_CONCATENATE,
            "link_start_indexes": cls.DROP,
            "links": cls.DROP,  # -- deprecated field --
            "email_message_id": cls.FIRST,
            "max_characters": cls.DROP,  # -- unused, remove from ElementMetadata --
            "orig_elements": cls.DROP,  # -- not expected, added by chunking, not before --
            "page_name": cls.FIRST,
            "page_number": cls.FIRST,
            "parent_id": cls.DROP,
            "routing": cls.DROP,
            "routing_score": cls.DROP,
            "sent_from": cls.FIRST,
            "sent_to": cls.FIRST,
            "signature": cls.FIRST,
            "subject": cls.FIRST,
            "text_as_html": cls.STRING_CONCATENATE,
            "table_as_cells": cls.FIRST,  # -- only occurs in Table --
            "table_extraction_method": cls.FIRST,
            "table_id": cls.DROP,  # -- added by chunking, not before --
            "chunk_index": cls.DROP,  # -- added by chunking, not before --
            "num_carried_over_header_rows": cls.DROP,  # -- added by chunking, not before --
            "url": cls.FIRST,
            # TODO: ideally a chunk spanning multiple audio segments would keep min(start) and
            # max(end) across its constituent elements. ConsolidationStrategy currently has no
            # MIN/MAX variants, so DROP is the safe fallback for now. Add MIN/MAX strategies
            # and switch these to cls.MIN / cls.MAX when that work is done.
            "segment_start_seconds": cls.DROP,
            "segment_end_seconds": cls.DROP,
            "key_value_pairs": cls.DROP,  # -- only occurs in FormKeysValues --
        }


_P = ParamSpec("_P")


def assign_and_map_hash_ids(elements: list[Element]) -> list[Element]:
    """Converts `id` and `parent_id` of elements from UUIDs to hashes.

    This function ensures deterministic IDs by:
    1. Converting each element's UUID into a hash.
    2. Updating the `parent_id` to match the new hash ID of parent elements.

    Args:
        elements: A list of Element objects to update.

    Returns:
        List of updated Element objects with hashes for `id` and `parent_id`.
    """
    # -- generate sequence number for each element on a page --
    page_numbers = [e.metadata.page_number for e in elements]
    page_seq_pairs = [
        seq_on_page for _, group in groupby(page_numbers) for seq_on_page, _ in enumerate(group)
    ]

    # -- assign hash IDs to elements --
    old_to_new_mapping = {
        element.id: element.id_to_hash(seq_on_page_counter)
        for element, seq_on_page_counter in zip(elements, page_seq_pairs)
    }

    # -- map old parent IDs to new ones --
    for e in elements:
        parent_id = e.metadata.parent_id
        if not parent_id or parent_id not in old_to_new_mapping:
            continue
        e.metadata.parent_id = old_to_new_mapping[parent_id]

    return elements


def process_metadata() -> Callable[[Callable[_P, list[Element]]], Callable[_P, list[Element]]]:
    """Post-process element-metadata for this document.

    This decorator adds a post-processing step to a document partitioner.

    - Adds `metadata_filename` parameter to docstring if not present.
    - Updates element.id to a UUID when `unique_element_ids` argument is provided and True.

    """

    def decorator(func: Callable[_P, list[Element]]) -> Callable[_P, list[Element]]:
        if func.__doc__:  # noqa: SIM102
            if (
                "metadata_filename" in func.__code__.co_varnames
                and "metadata_filename" not in func.__doc__
            ):
                func.__doc__ += (
                    "\nMetadata Parameters:\n\tmetadata_filename:"
                    + "\n\t\tThe filename to use in element metadata."
                )

        @functools.wraps(func)
        def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> list[Element]:
            elements = func(*args, **kwargs)
            call_args = get_call_args_applying_defaults(func, *args, **kwargs)

            unique_element_ids: bool = call_args.get("unique_element_ids", False)
            if unique_element_ids is False:
                elements = assign_and_map_hash_ids(elements)

            return elements

        return wrapper

    return decorator


class ElementType:
    TITLE = "Title"
    TEXT = "Text"
    UNCATEGORIZED_TEXT = "UncategorizedText"
    NARRATIVE_TEXT = "NarrativeText"
    BULLETED_TEXT = "BulletedText"
    PARAGRAPH = "Paragraph"
    ABSTRACT = "Abstract"
    THREADING = "Threading"
    FORM = "Form"
    FIELD_NAME = "Field-Name"
    VALUE = "Value"
    LINK = "Link"
    COMPOSITE_ELEMENT = "CompositeElement"
    IMAGE = "Image"
    PICTURE = "Picture"
    FIGURE_CAPTION = "FigureCaption"
    FIGURE = "Figure"
    CAPTION = "Caption"
    LIST = "List"
    LIST_ITEM = "ListItem"
    LIST_ITEM_OTHER = "List-item"
    CHECKED = "Checked"
    UNCHECKED = "Unchecked"
    CHECK_BOX_CHECKED = "CheckBoxChecked"
    CHECK_BOX_UNCHECKED = "CheckBoxUnchecked"
    RADIO_BUTTON_CHECKED = "RadioButtonChecked"
    RADIO_BUTTON_UNCHECKED = "RadioButtonUnchecked"
    ADDRESS = "Address"
    EMAIL_ADDRESS = "EmailAddress"
    PAGE_BREAK = "PageBreak"
    FORMULA = "Formula"
    TABLE = "Table"
    HEADER = "Header"
    HEADLINE = "Headline"
    SUB_HEADLINE = "Subheadline"
    PAGE_HEADER = "Page-header"  # Title?
    SECTION_HEADER = "Section-header"
    FOOTER = "Footer"
    FOOTNOTE = "Footnote"
    PAGE_FOOTER = "Page-footer"
    PAGE_NUMBER = "PageNumber"
    CODE_SNIPPET = "CodeSnippet"
    FORM_KEYS_VALUES = "FormKeysValues"
    DOCUMENT_DATA = "DocumentData"

    @classmethod
    def to_dict(cls):
        """
        Convert class attributes to a dictionary.

        Returns:
            dict: A dictionary where keys are attribute names and values are attribute values.
        """
        return {
            attr: getattr(cls, attr)
            for attr in dir(cls)
            if not callable(getattr(cls, attr)) and not attr.startswith("__")
        }


class Element(abc.ABC):
    """An element is a semantically-coherent component of a document, often a paragraph.

    There are a few design principles that are followed when creating an element:
    1. It will always have an ID, which by default is a random UUID.
    2. Asking for an ID should always return a string, it can never be None.
    3. ID is lazy, meaning it will be generated when asked for the first time.
    4. When deterministic behavior is needed, the ID can be converted.
        to a hash based on its text `element.id_to_hash(position)`
    4. Even if the `text` attribute is not defined in a subclass, it will default to a blank string.
    6. Assigning a string ID manually is possible, but is meant to be used
        only for deserialization purposes.
    """

    text: str
    category = "UncategorizedText"

    def __init__(
        self,
        element_id: Optional[str] = None,
        coordinates: Optional[tuple[tuple[float, float], ...]] = None,
        coordinate_system: Optional[CoordinateSystem] = None,
        metadata: Optional[ElementMetadata] = None,
        detection_origin: Optional[str] = None,
    ):
        if element_id is not None and not isinstance(element_id, str):  # type: ignore
            raise ValueError("element_id must be of type str or None.")

        self._element_id = element_id
        self.metadata = ElementMetadata() if metadata is None else metadata
        if coordinates is not None or coordinate_system is not None:
            self.metadata.coordinates = CoordinatesMetadata(
                points=coordinates, system=coordinate_system
            )
        self.metadata.detection_origin = detection_origin
        # -- all `Element` instances get a `text` attribute, defaults to the empty string if not
        # -- defined in a subclass.
        self.text = self.text if hasattr(self, "text") else ""

    def __str__(self):
        return self.text

    def convert_coordinates_to_new_system(
        self, new_system: CoordinateSystem, in_place: bool = True
    ) -> Optional[Points]:
        """Converts the element location coordinates to a new coordinate system.

        If inplace is true, changes the coordinates in place and updates the coordinate system.
        """
        if (
            self.metadata.coordinates is None
            or self.metadata.coordinates.system is None
            or self.metadata.coordinates.points is None
        ):
            return None

        new_coordinates = tuple(
            self.metadata.coordinates.system.convert_coordinates_to_new_system(
                new_system=new_system,
                x=x,
                y=y,
            )
            for x, y in self.metadata.coordinates.points
        )

        if in_place:
            self.metadata.coordinates.points = new_coordinates
            self.metadata.coordinates.system = new_system

        return new_coordinates

    def id_to_hash(self, sequence_number: int) -> str:
        """Calculates and assigns a deterministic hash as an ID.

        The hash ID is based on element's text, sequence number on page,
        page number and its filename.

        Args:
            sequence_number: index on page

        Returns: new ID value
        """
        data = f"{self.metadata.filename}{self.text}{self.metadata.page_number}{sequence_number}"
        self._element_id = hashlib.sha256(data.encode()).hexdigest()[:32]
        return self.id

    @property
    def id(self):
        if self._element_id is None:
            self._element_id = str(uuid.uuid4())
        return self._element_id

    def to_dict(self) -> dict[str, Any]:
        return {
            "type": None,
            "element_id": self.id,
            "text": self.text,
            "metadata": self.metadata.to_dict(),
        }


class CheckBox(Element):
    """A checkbox with an attribute indicating whether its checked or not.

    Primarily used in documents that are forms.
    """

    def __init__(
        self,
        element_id: Optional[str] = None,
        coordinates: Optional[tuple[tuple[float, float], ...]] = None,
        coordinate_system: Optional[CoordinateSystem] = None,
        checked: bool = False,
        metadata: Optional[ElementMetadata] = None,
        detection_origin: Optional[str] = None,
    ):
        metadata = metadata if metadata else ElementMetadata()
        super().__init__(
            element_id=element_id,
            coordinates=coordinates,
            coordinate_system=coordinate_system,
            metadata=metadata,
            detection_origin=detection_origin,
        )
        self.checked: bool = checked

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, CheckBox):
            return False
        return all(
            (
                self.checked == other.checked,
                self.metadata.coordinates == other.metadata.coordinates,
            )
        )

    def to_dict(self) -> dict[str, Any]:
        """Serialize to JSON-compatible (str keys) dict."""
        out = super().to_dict()
        out["type"] = "CheckBox"
        out["checked"] = self.checked
        out["element_id"] = self.id
        return out


class Text(Element):
    """Base element for capturing free text from within document."""

    def __init__(
        self,
        text: str,
        element_id: Optional[str] = None,
        coordinates: Optional[tuple[tuple[float, float], ...]] = None,
        coordinate_system: Optional[CoordinateSystem] = None,
        metadata: Optional[ElementMetadata] = None,
        detection_origin: Optional[str] = None,
        embeddings: Optional[list[float]] = None,
    ):
        metadata = metadata if metadata else ElementMetadata()
        self.text: str = text
        self.embeddings: Optional[list[float]] = embeddings

        super().__init__(
            element_id=element_id,
            metadata=metadata,
            coordinates=coordinates,
            coordinate_system=coordinate_system,
            detection_origin=detection_origin,
        )

    def __eq__(self, other: object):
        if not isinstance(other, Text):
            return False
        return all(
            (
                self.text == other.text,
                self.metadata.coordinates == other.metadata.coordinates,
                self.category == other.category,
                self.embeddings == other.embeddings,
            ),
        )

    def __str__(self):
        return self.text

    def apply(self, *cleaners: Callable[[str], str]):
        """Applies a cleaning brick to the text element.

        The function that's passed in should take a string as input and produce a string as
        output.
        """
        cleaned_text = self.text
        for cleaner in cleaners:
            cleaned_text = cleaner(cleaned_text)

        if not isinstance(cleaned_text, str):  # pyright: ignore[reportUnnecessaryIsInstance]
            raise ValueError("Cleaner produced a non-string output.")

        self.text = cleaned_text

    def to_dict(self) -> dict[str, Any]:
        """Serialize to JSON-compatible (str keys) dict."""
        out = super().to_dict()
        out["element_id"] = self.id
        out["type"] = self.category
        out["text"] = self.text
        if self.embeddings:
            out["embeddings"] = self.embeddings
        return out


class Formula(Text):
    "An element containing formulas in a document"

    category = "Formula"


class CompositeElement(Text):
    """A chunk formed from text (non-Table) elements.

    Only produced by chunking. An instance may be formed by combining one or more sequential
    elements produced by partitioning. It it also used when text-splitting an "oversized" element,
    a single element that by itself is larger than the requested chunk size.
    """

    category = "CompositeElement"


class FigureCaption(Text):
    """An element for capturing text associated with figure captions."""

    category = "FigureCaption"


class NarrativeText(Text):
    """NarrativeText is an element consisting of multiple, well-formulated sentences. This
    excludes elements such titles, headers, footers, and captions."""

    category = "NarrativeText"


class Form(Text):
    """An element for capturing form text."""

    category = "Form"


class ListItem(Text):
    """ListItem is a NarrativeText element that is part of a list."""

    category = "ListItem"


class Title(Text):
    """A text element for capturing titles."""

    category = "Title"


class Address(Text):
    """A text element for capturing addresses."""

    category = "Address"


class EmailAddress(Text):
    """A text element for capturing addresses"""

    category = "EmailAddress"


class Image(Text):
    """A text element for capturing image metadata."""

    category = ElementType.IMAGE


class PageBreak(Text):
    """An element for capturing page breaks."""

    category = "PageBreak"


class Table(Text):
    """An element for capturing tables."""

    category = "Table"


class TableChunk(Table):
    """An element for capturing chunks of tables."""

    category = "TableChunk"


class Header(Text):
    """An element for capturing document headers."""

    category = "Header"


class Footer(Text):
    """An element for capturing document footers."""

    category = "Footer"


class CodeSnippet(Text):
    """An element for capturing code snippets."""

    category = "CodeSnippet"


class PageNumber(Text):
    """An element for capturing page numbers."""

    category = "PageNumber"


class FormKeysValues(Text):
    """An element for capturing Key-Value dicts (forms)."""

    category = "FormKeysValues"


class DocumentData(Text):
    """An element for capturing document-level data,
    particularly for large data that does not make sense to
    represent across each element in the document."""

    category = "DocumentData"


TYPE_TO_TEXT_ELEMENT_MAP: dict[str, type[Text]] = {
    ElementType.TITLE: Title,
    ElementType.SECTION_HEADER: Title,
    ElementType.HEADLINE: Title,
    ElementType.SUB_HEADLINE: Title,
    ElementType.FIELD_NAME: Title,
    ElementType.UNCATEGORIZED_TEXT: Text,
    ElementType.COMPOSITE_ELEMENT: CompositeElement,
    ElementType.TEXT: NarrativeText,
    ElementType.NARRATIVE_TEXT: NarrativeText,
    ElementType.PARAGRAPH: NarrativeText,
    # this mapping favors ensures yolox produces backward compatible categories
    ElementType.ABSTRACT: NarrativeText,
    ElementType.THREADING: NarrativeText,
    ElementType.FORM: Form,
    ElementType.VALUE: NarrativeText,
    ElementType.LINK: NarrativeText,
    ElementType.LIST_ITEM: ListItem,
    ElementType.BULLETED_TEXT: ListItem,
    ElementType.LIST_ITEM_OTHER: ListItem,
    ElementType.HEADER: Header,
    ElementType.PAGE_HEADER: Header,  # Title?
    ElementType.FOOTER: Footer,
    ElementType.PAGE_FOOTER: Footer,
    ElementType.FOOTNOTE: Footer,
    ElementType.FIGURE_CAPTION: FigureCaption,
    ElementType.CAPTION: FigureCaption,
    ElementType.IMAGE: Image,
    ElementType.FIGURE: Image,
    ElementType.PICTURE: Image,
    ElementType.TABLE: Table,
    ElementType.ADDRESS: Address,
    ElementType.EMAIL_ADDRESS: EmailAddress,
    ElementType.FORMULA: Formula,
    ElementType.PAGE_BREAK: PageBreak,
    ElementType.CODE_SNIPPET: CodeSnippet,
    ElementType.PAGE_NUMBER: PageNumber,
    ElementType.FORM_KEYS_VALUES: FormKeysValues,
    ElementType.DOCUMENT_DATA: DocumentData,
}


def _kvform_rehydrate_internal_elements(kv_pairs: list[dict[str, Any]]) -> list[FormKeyValuePair]:
    """
    The key_value_pairs metadata field contains (in the vast majority of cases)
    nested Text elements. Those need to be turned from dicts into Elements explicitly,
    e.g. when partition_json is used.
    """
    from unstructured.staging.base import elements_from_dicts

    # safe to overwrite - deepcopy already happened
    for kv_pair in kv_pairs:
        if kv_pair["key"]["custom_element"] is not None:
            (kv_pair["key"]["custom_element"],) = elements_from_dicts(
                [kv_pair["key"]["custom_element"]]
            )
        if kv_pair["value"] is not None and kv_pair["value"]["custom_element"] is not None:
            (kv_pair["value"]["custom_element"],) = elements_from_dicts(
                [kv_pair["value"]["custom_element"]]
            )
    return cast(list[FormKeyValuePair], kv_pairs)


def _kvform_pairs_to_dict(orig_kv_pairs: list[FormKeyValuePair]) -> list[dict[str, Any]]:
    """
    The key_value_pairs metadata field contains (in the vast majority of cases)
    nested Text elements. Those need to be turned from Elements to dicts recursively,
    e.g. when FormKeysValues.to_dict() is used.

    """
    kv_pairs: list[dict[str, Any]] = copy.deepcopy(orig_kv_pairs)  # type: ignore
    for kv_pair in kv_pairs:
        if kv_pair["key"]["custom_element"] is not None:
            kv_pair["key"]["custom_element"] = kv_pair["key"]["custom_element"].to_dict()
        if kv_pair["value"] is not None and kv_pair["value"]["custom_element"] is not None:
            kv_pair["value"]["custom_element"] = kv_pair["value"]["custom_element"].to_dict()

    return kv_pairs