import csv
import json
import logging
import platform
import re
import sys
import tarfile
import zipfile
from collections.abc import Iterable, Mapping
from datetime import datetime
from enum import Enum
from io import BytesIO
from pathlib import Path, PurePath
from typing import (
    TYPE_CHECKING,
    Annotated,
    Literal,
    NamedTuple,
    Optional,
    Type,
    Union,
)

import filetype

# DO NOT REMOVE; explicitly exposed from this location
from docling_core.types.doc import (
    DocItem,
    DocItemLabel,
    DoclingDocument,
    PictureItem,
    SectionHeaderItem,
    TableItem,
    TextItem,
)
from docling_core.utils.file import (
    FileSizeLimitExceededError,
    resolve_remote_filename,
    resolve_source_to_stream,
)
from pydantic import (
    AnyHttpUrl,
    BaseModel,
    Field,
    PrivateAttr,
    TypeAdapter,
    ValidationError,
)

from docling.backend.abstract_backend import (
    AbstractDocumentBackend,
    PaginatedDocumentBackend,
)
from docling.datamodel.backend_options import (
    BackendOptions,
    MetsGbsBackendOptions,
)
from docling.datamodel.base_models import (
    AssembledUnit,
    ConfidenceReport,
    ConversionStatus,
    DoclingComponentType,
    DocumentStream,
    ErrorItem,
    FailureCategory,
    FormatToExtensions,
    FormatToMimeType,
    InputFormat,
    MimeTypeToFormat,
    Page,
)
from docling.datamodel.settings import DocumentLimits
from docling.exceptions import DocumentLoadError
from docling.utils.profiling import ProfilingItem
from docling.utils.utils import create_file_hash, safe_version

if TYPE_CHECKING:
    from docling.datamodel.base_models import BaseFormatOption

_log = logging.getLogger(__name__)

layout_label_to_ds_type = {
    DocItemLabel.TITLE: "title",
    DocItemLabel.DOCUMENT_INDEX: "table",
    DocItemLabel.SECTION_HEADER: "subtitle-level-1",
    DocItemLabel.CHECKBOX_SELECTED: "checkbox-selected",
    DocItemLabel.CHECKBOX_UNSELECTED: "checkbox-unselected",
    DocItemLabel.CAPTION: "caption",
    DocItemLabel.PAGE_HEADER: "page-header",
    DocItemLabel.PAGE_FOOTER: "page-footer",
    DocItemLabel.FOOTNOTE: "footnote",
    DocItemLabel.TABLE: "table",
    DocItemLabel.FORMULA: "equation",
    DocItemLabel.LIST_ITEM: "paragraph",
    DocItemLabel.CODE: "paragraph",
    DocItemLabel.PICTURE: "figure",
    DocItemLabel.TEXT: "paragraph",
    DocItemLabel.PARAGRAPH: "paragraph",
    DocItemLabel.FORM: DocItemLabel.FORM.value,
    DocItemLabel.KEY_VALUE_REGION: DocItemLabel.KEY_VALUE_REGION.value,
}

_EMPTY_DOCLING_DOC = DoclingDocument(name="dummy")


class InputRejection(NamedTuple):
    """Why an input document was flagged invalid, captured where it is known.

    Set on ``InputDocument`` in ``__init__`` and ``create_invalid`` so the
    converter can emit a categorized ``ErrorItem`` instead of an empty errors
    list. Transient plumbing (a private attr), not part of the serialized model.
    """

    message: str
    category: FailureCategory


class InputDocument(BaseModel):
    """A document as an input of a Docling conversion."""

    file: Annotated[
        PurePath, Field(description="A path representation the input document.")
    ]
    document_hash: Annotated[
        str,
        Field(description="A stable hash of the path or stream of the input document."),
    ]
    valid: bool = Field(True, description="Whether this is is a valid input document.")
    backend_options: Optional[BackendOptions] = Field(
        None, description="Custom options for backends."
    )
    limits: DocumentLimits = Field(
        DocumentLimits(), description="Limits in the input document for the conversion."
    )
    format: Annotated[InputFormat, Field(description="The document format.")]

    filesize: Optional[int] = Field(
        None, description="Size of the input file, in bytes."
    )
    page_count: int = Field(0, description="Number of pages in the input document.")

    _backend: AbstractDocumentBackend
    # Reason this input was flagged invalid, if any (transient, not serialized).
    _rejection: Optional[InputRejection] = PrivateAttr(default=None)

    def __init__(
        self,
        path_or_stream: Union[BytesIO, Path],
        format: InputFormat,
        backend: Type[AbstractDocumentBackend],
        backend_options: Optional[BackendOptions] = None,
        filename: Optional[str] = None,
        limits: Optional[DocumentLimits] = None,
    ) -> None:
        super().__init__(
            file="",
            document_hash="",
            format=InputFormat.PDF,
            backend_options=backend_options,
        )  # initialize with dummy values
        self.limits = limits or DocumentLimits()
        self.format = format

        try:
            if isinstance(path_or_stream, Path):
                self.file = path_or_stream
                self.filesize = path_or_stream.stat().st_size
                if self.filesize > self.limits.max_file_size:
                    self._reject_filesize()
                else:
                    self.document_hash = create_file_hash(path_or_stream)
                    self._init_doc(backend, path_or_stream)

            elif isinstance(path_or_stream, BytesIO):
                assert filename is not None, (
                    "Can't construct InputDocument from stream without providing "
                    "filename arg."
                )
                self.file = PurePath(filename)
                self.filesize = path_or_stream.getbuffer().nbytes

                if self.filesize > self.limits.max_file_size:
                    self._reject_filesize()
                else:
                    self.document_hash = create_file_hash(path_or_stream)
                    self._init_doc(backend, path_or_stream)
            else:
                raise RuntimeError(
                    f"Unexpected type path_or_stream: {type(path_or_stream)}"
                )

            # For paginated backends, check if the maximum page count is exceeded.
            if self.valid and self._backend.is_valid():
                if self._backend.supports_pagination() and isinstance(
                    self._backend, PaginatedDocumentBackend
                ):
                    self.page_count = self._backend.page_count()
                    if not self.page_count <= self.limits.max_num_pages:
                        self.valid = False
                        self._rejection = InputRejection(
                            message=(
                                f"Document has {self.page_count} pages, exceeding the "
                                f"max_num_pages limit of {self.limits.max_num_pages}."
                            ),
                            category=FailureCategory.POLICY,
                        )
                    elif self.page_count < self.limits.page_range[0]:
                        self.valid = False
                        self._rejection = InputRejection(
                            message=(
                                f"Document has {self.page_count} pages, fewer than the "
                                f"requested page_range start {self.limits.page_range[0]}."
                            ),
                            category=FailureCategory.POLICY,
                        )

        except (FileNotFoundError, OSError) as e:
            self.valid = False
            self._rejection = self._rejection or InputRejection(
                message=f"File {self.file.name} not found or cannot be opened.",
                category=FailureCategory.SOURCE_UNAVAILABLE,
            )
            _log.exception(
                f"File {self.file.name} not found or cannot be opened.", exc_info=e
            )
        except RuntimeError as e:
            # Local RuntimeErrors (e.g. the "Unexpected type" guard) that aren't a
            # backend bad-input signal; categorized UNKNOWN. (Backend parse
            # failures are handled in _init_doc and don't reach here.)
            self.valid = False
            self._rejection = self._rejection or InputRejection(
                message=(
                    "An unexpected error occurred while opening the document "
                    f"{self.file.name}."
                ),
                category=FailureCategory.UNKNOWN,
            )
            _log.exception(
                "An unexpected error occurred while opening the document "
                f"{self.file.name}",
                exc_info=e,
            )

    @classmethod
    def create_invalid(
        cls,
        *,
        filename: str,
        format: InputFormat,
        filesize: int,
        limits: Optional[DocumentLimits] = None,
        rejection: Optional[InputRejection] = None,
    ) -> "InputDocument":
        """Build an InputDocument flagged invalid without opening a backend.

        Used when the input is rejected before a stream is available, e.g. an
        HTTP download aborted for exceeding ``limits.max_file_size``. The normal
        constructor derives ``filesize`` from the actual path/stream, which is
        unavailable in that case, so the fields are set explicitly here.

        ``__init__`` is overridden to load from a path/stream and open a backend,
        so the validated constructor cannot be used to set bare field values;
        ``model_construct`` is the supported way to do that. Only fields that
        differ from their declared defaults are passed; the rest (e.g.
        ``backend_options``, ``page_count``) fall back to those defaults.

        ``rejection`` carries the (message, category) reason so the converter can
        emit a categorized ``ErrorItem`` instead of an empty errors list.
        """
        doc = cls.model_construct(
            file=PurePath(filename),
            document_hash="",
            valid=False,
            limits=limits or DocumentLimits(),
            format=format,
            filesize=filesize,
        )
        doc._rejection = rejection
        return doc

    def _reject_filesize(self) -> None:
        self.valid = False
        self._rejection = InputRejection(
            message=(
                f"File size {self.filesize} exceeds the max_file_size limit of "
                f"{self.limits.max_file_size} bytes."
            ),
            category=FailureCategory.POLICY,
        )

    def _init_doc(
        self,
        backend: Type[AbstractDocumentBackend],
        path_or_stream: Union[BytesIO, Path],
    ) -> None:
        try:
            if self.backend_options:
                self._backend = backend(
                    self,
                    path_or_stream=path_or_stream,
                    options=self.backend_options,
                )
            else:
                self._backend = backend(self, path_or_stream=path_or_stream)
        except Exception as exc:
            # A DocumentLoadError (bad input bytes) is recorded as a
            # BACKEND_FAILURE rejection, like the is_valid() branch below.
            # Anything else (missing dependency, bug) propagates so an internal
            # defect is not mislabeled.
            if not isinstance(exc, DocumentLoadError):
                raise
            self.valid = False
            self._rejection = InputRejection(
                message=str(exc) or "The document backend could not parse the input.",
                category=FailureCategory.BACKEND_FAILURE,
            )
            return

        if not self._backend.is_valid():
            self.valid = False
            self._rejection = InputRejection(
                message="The document backend could not parse the input.",
                category=FailureCategory.BACKEND_FAILURE,
            )


def build_invalid_input_errors(in_doc: "InputDocument") -> list[ErrorItem]:
    """Build the ErrorItem list for an invalid input document.

    Surfaces the rejection reason captured during construction so the document
    reaches the user as a categorized error instead of an empty errors list.
    Falls back to a generic UNKNOWN entry if no reason was recorded.
    """
    rejection = in_doc._rejection
    if rejection is None:
        return [
            ErrorItem(
                component_type=DoclingComponentType.USER_INPUT,
                module_name="",
                error_message="Input document is not valid.",
                category=FailureCategory.UNKNOWN,
            )
        ]
    return [
        ErrorItem(
            component_type=DoclingComponentType.USER_INPUT,
            module_name="",
            error_message=rejection.message,
            category=rejection.category,
        )
    ]


class DocumentFormat(str, Enum):
    V2 = "v2"
    V1 = "v1"


class DoclingVersion(BaseModel):
    docling_version: str = safe_version("docling")
    docling_slim_version: str = safe_version("docling-slim")
    docling_core_version: str = safe_version("docling-core")
    docling_ibm_models_version: str = safe_version("docling-ibm-models")
    docling_parse_version: str = safe_version("docling-parse")
    platform_str: str = platform.platform()
    py_impl_version: str = sys.implementation.cache_tag
    py_lang_version: str = platform.python_version()


class ConversionAssets(BaseModel):
    version: DoclingVersion = DoclingVersion()
    # When the assets were saved (ISO string from datetime.now())
    timestamp: Optional[str] = None

    status: ConversionStatus = ConversionStatus.PENDING  # failure, success
    errors: list[ErrorItem] = []  # structure to keep errors

    pages: list[Page] = []
    timings: dict[str, ProfilingItem] = {}
    confidence: ConfidenceReport = Field(default_factory=ConfidenceReport)

    document: DoclingDocument = _EMPTY_DOCLING_DOC

    def has_errors(self, category: Optional[FailureCategory] = None) -> bool:
        """Whether any error was recorded.

        Args:
            category: If given, only errors of this category count; otherwise
                any recorded error qualifies.
        """
        if category is None:
            return bool(self.errors)
        return any(e.category == category for e in self.errors)

    def has_timeout_errors(self) -> bool:
        """Whether any error has category TIMEOUT."""
        return self.has_errors(FailureCategory.TIMEOUT)

    def has_inference_errors(self) -> bool:
        """Whether any error has category INFERENCE_FAILURE."""
        return self.has_errors(FailureCategory.INFERENCE_FAILURE)

    def has_parse_errors(self) -> bool:
        """Whether any error has category BACKEND_FAILURE (backend/parse failure)."""
        return self.has_errors(FailureCategory.BACKEND_FAILURE)

    def save(
        self,
        *,
        filename: Union[str, Path],
        indent: Optional[int] = 2,
    ):
        """Serialize the full ConversionAssets to JSON."""
        if isinstance(filename, str):
            filename = Path(filename)
        # Build an in-memory ZIP archive containing JSON for each asset
        buf = BytesIO()

        def to_jsonable(obj):
            try:
                # pydantic v2 models
                if hasattr(obj, "model_dump"):
                    return obj.model_dump(mode="json")  # type: ignore[attr-defined]
            except TypeError:
                # some models may not accept mode argument
                return obj.model_dump()  # type: ignore[attr-defined]

            # enums
            try:
                from enum import Enum

                if isinstance(obj, Enum):
                    return obj.value
            except Exception:
                pass

            # containers
            if isinstance(obj, list):
                return [to_jsonable(x) for x in obj]
            if isinstance(obj, dict):
                return {k: to_jsonable(v) for k, v in obj.items()}

            # passthrough primitives
            return obj

        with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:

            def write_json(name: str, payload) -> None:
                data = json.dumps(
                    to_jsonable(payload), ensure_ascii=False, indent=indent
                )
                zf.writestr(name, data.encode("utf-8"))

            # Update and persist a save timestamp
            self.timestamp = datetime.now().isoformat()
            write_json("timestamp.json", self.timestamp)

            # Store each component in its own JSON file
            write_json("version.json", self.version)
            write_json("status.json", self.status)
            write_json("errors.json", self.errors)
            write_json("pages.json", self.pages)
            write_json("timings.json", self.timings)
            write_json("confidence.json", self.confidence)
            # For the document, ensure stable schema via export_to_dict
            doc_dict = self.document.export_to_dict()
            zf.writestr(
                "document.json",
                json.dumps(doc_dict, ensure_ascii=False, indent=indent).encode("utf-8"),
            )

        # Persist the ZIP to disk
        buf.seek(0)
        if filename.parent and not filename.parent.exists():
            filename.parent.mkdir(parents=True, exist_ok=True)
        with filename.open("wb") as f:
            f.write(buf.getvalue())

    @classmethod
    def load(cls, filename: Union[str, Path]) -> "ConversionAssets":
        """Load a ConversionAssets."""
        if isinstance(filename, str):
            filename = Path(filename)

        # Read the ZIP and deserialize all items
        version_info: DoclingVersion = DoclingVersion()
        timestamp: Optional[str] = None
        status = ConversionStatus.PENDING
        errors: list[ErrorItem] = []
        pages: list[Page] = []
        timings: dict[str, ProfilingItem] = {}
        confidence = ConfidenceReport()
        document: DoclingDocument = _EMPTY_DOCLING_DOC

        with zipfile.ZipFile(filename, mode="r") as zf:

            def read_json(name: str):
                try:
                    with zf.open(name, "r") as fp:
                        return json.loads(fp.read().decode("utf-8"))
                except KeyError:
                    return None

            # version
            if (data := read_json("version.json")) is not None:
                try:
                    version_info = DoclingVersion.model_validate(data)
                except Exception as exc:
                    _log.error(f"Could not read version: {exc}")

            # timestamp
            if (data := read_json("timestamp.json")) is not None:
                if isinstance(data, str):
                    timestamp = data

            # status
            if (data := read_json("status.json")) is not None:
                try:
                    status = ConversionStatus(data)
                except Exception:
                    status = ConversionStatus.PENDING

            # errors
            if (data := read_json("errors.json")) is not None and isinstance(
                data, list
            ):
                errors = [ErrorItem.model_validate(item) for item in data]

            # pages
            if (data := read_json("pages.json")) is not None and isinstance(data, list):
                pages = [Page.model_validate(item) for item in data]

            # timings
            if (data := read_json("timings.json")) is not None and isinstance(
                data, dict
            ):
                timings = {k: ProfilingItem.model_validate(v) for k, v in data.items()}

            # confidence
            if (data := read_json("confidence.json")) is not None and isinstance(
                data, dict
            ):
                confidence = ConfidenceReport.model_validate(data)

            # document
            if (data := read_json("document.json")) is not None and isinstance(
                data, dict
            ):
                document = DoclingDocument.model_validate(data)

        return cls(
            version=version_info,
            timestamp=timestamp,
            status=status,
            errors=errors,
            pages=pages,
            timings=timings,
            confidence=confidence,
            document=document,
        )


class ConversionResult(ConversionAssets):
    input: InputDocument
    assembled: AssembledUnit = AssembledUnit()


class _DummyBackend(AbstractDocumentBackend):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def is_valid(self) -> bool:
        return False

    @classmethod
    def supported_formats(cls) -> set[InputFormat]:
        return set()

    @classmethod
    def supports_pagination(cls) -> bool:
        return False

    def unload(self):
        return super().unload()


class _DocumentConversionInput(BaseModel):
    path_or_stream_iterator: Iterable[Union[Path, str, DocumentStream]]
    headers: Optional[dict[str, str]] = None
    limits: Optional[DocumentLimits] = DocumentLimits()

    def docs(
        self,
        format_options: Mapping[InputFormat, "BaseFormatOption"],
    ) -> Iterable[InputDocument]:
        for item in self.path_or_stream_iterator:
            if isinstance(item, str):
                try:
                    obj = resolve_source_to_stream(
                        item,
                        self.headers,
                        max_file_size=self.limits.max_file_size,
                    )
                except FileSizeLimitExceededError as exc:
                    yield self._build_invalid_input_document(
                        name=exc.filename,
                        format_options=format_options,
                        file_size=exc.size,
                        rejection=InputRejection(
                            message=(
                                f"File size {exc.size} exceeds the max_file_size "
                                f"limit of {self.limits.max_file_size} bytes."
                            ),
                            category=FailureCategory.POLICY,
                        ),
                    )
                    continue
                except (OSError, ValueError) as exc:
                    # A source that cannot be fetched or resolved -- unreachable
                    # URL, HTTP error status, connection/timeout failure, unsafe or
                    # malformed URL, missing local file, ... -- must not abort the
                    # whole batch. Emit an invalid InputDocument so it surfaces as a
                    # document-level FAILURE that still honors raises_on_error
                    # (i.e. aborts only when abort_on_error is set). requests'
                    # RequestException subclasses derive from OSError, so this also
                    # covers all HTTP fetch errors without importing requests here.
                    _log.error("Failed to resolve input source %r: %s", item, exc)
                    yield self._build_invalid_input_document(
                        name=self._filename_from_source(item),
                        format_options=format_options,
                        rejection=self._classify_source_error(exc),
                    )
                    continue
            else:
                obj = item
            format = self._guess_format(obj)
            backend: Type[AbstractDocumentBackend]
            backend_options: Optional[BackendOptions] = None
            if not format or format not in format_options:
                _log.error(
                    f"Input document {obj.name} with format {format} does not match "
                    f"any allowed format: ({format_options.keys()})"
                )
                backend = _DummyBackend
            else:
                options = format_options[format]
                backend = options.backend
                backend_options = options.backend_options_for_input(item)

            path_or_stream: Union[BytesIO, Path]
            if isinstance(obj, Path):
                path_or_stream = obj
            elif isinstance(obj, DocumentStream):
                path_or_stream = obj.stream
            else:
                raise RuntimeError(f"Unexpected obj type in iterator: {type(obj)}")

            yield InputDocument(
                path_or_stream=path_or_stream,
                format=format,  # type: ignore[arg-type]
                filename=obj.name,
                limits=self.limits,
                backend=backend,
                backend_options=backend_options,
            )

    def _build_invalid_input_document(
        self,
        name: str,
        format_options: Mapping[InputFormat, "BaseFormatOption"],
        file_size: int = 0,
        rejection: Optional[InputRejection] = None,
    ) -> InputDocument:
        guessed_format = self._guess_format(DocumentStream(name=name, stream=BytesIO()))
        if guessed_format is None:
            guessed_format = next(iter(format_options.keys()))

        return InputDocument.create_invalid(
            filename=name,
            format=guessed_format,
            filesize=file_size,
            limits=self.limits,
            rejection=rejection,
        )

    @staticmethod
    def _classify_source_error(exc: BaseException) -> InputRejection:
        """Map a source-resolution failure to an InputRejection.

        Splits HTTP status rejections the same way as the jobkit task path
        (``_classify_http_status``): policy status codes map to POLICY, every
        other status or transport failure maps to SOURCE_UNAVAILABLE.
        ``response.status_code`` is read by duck-typing to avoid a ``requests``
        import.
        """
        response = getattr(exc, "response", None)
        status_code = getattr(response, "status_code", None)
        if status_code in {401, 403, 404, 413, 415, 422}:
            return InputRejection(message=str(exc), category=FailureCategory.POLICY)
        return InputRejection(
            message=str(exc) or "Source document could not be reached.",
            category=FailureCategory.SOURCE_UNAVAILABLE,
        )

    @staticmethod
    def _filename_from_source(source: str) -> str:
        """Best-effort filename for a source that could not be resolved.

        Reuses docling-core's ``resolve_remote_filename`` for URLs (the same
        helper ``resolve_source_to_stream`` uses to name successful fetches) so a
        failed source is labeled consistently; the URL path basename is used with
        the query string dropped. Non-URL sources fall back to the path basename.
        """
        try:
            http_url = TypeAdapter(AnyHttpUrl).validate_python(source)
        except ValidationError:
            return PurePath(source).name or source
        return resolve_remote_filename(http_url=http_url, response_headers={})

    def _guess_format(self, obj: Union[Path, DocumentStream]) -> Optional[InputFormat]:
        content = b""  # empty binary blob
        formats: list[InputFormat] = []
        obj_ext: Optional[str] = None

        if isinstance(obj, Path):
            if _DocumentConversionInput._has_doclang_extension(obj.name):
                return InputFormat.XML_DOCLANG
            mime = filetype.guess_mime(str(obj))
            obj_ext = obj.suffix[1:] if obj.suffix else ""
            if mime is None:
                mime = _DocumentConversionInput._mime_from_extension(obj_ext)
            if mime is None:  # must guess from content
                with obj.open("rb") as f:
                    content = f.read(1024)  # Read first 1KB
            if mime is not None and mime.lower() == "application/zip":
                mime_root = "application/vnd.openxmlformats-officedocument"
                suffix = obj.suffix.lower()
                if suffix == ".xlsx":
                    mime = mime_root + ".spreadsheetml.sheet"
                elif suffix == ".docx":
                    mime = mime_root + ".wordprocessingml.document"
                elif suffix == ".pptx":
                    mime = mime_root + ".presentationml.presentation"
                else:
                    office_mime = _DocumentConversionInput._detect_office_mime_from_zip(
                        obj
                    )
                    if office_mime is not None:
                        mime = office_mime

        elif isinstance(obj, DocumentStream):
            if _DocumentConversionInput._has_doclang_extension(obj.name):
                return InputFormat.XML_DOCLANG
            content = obj.stream.read(8192)
            obj.stream.seek(0)
            mime = filetype.guess_mime(content)
            obj_ext = (
                obj.name.rsplit(".", 1)[-1]
                if ("." in obj.name and not obj.name.startswith("."))
                else ""
            )
            if mime is None:
                mime = _DocumentConversionInput._mime_from_extension(obj_ext.lower())
            if mime is not None and mime.lower() == "application/zip":
                objname = obj.name.lower()
                mime_root = "application/vnd.openxmlformats-officedocument"
                if objname.endswith(".xlsx"):
                    mime = mime_root + ".spreadsheetml.sheet"
                elif objname.endswith(".docx"):
                    mime = mime_root + ".wordprocessingml.document"
                elif objname.endswith(".pptx"):
                    mime = mime_root + ".presentationml.presentation"
                else:
                    office_mime = _DocumentConversionInput._detect_office_mime_from_zip(
                        obj.stream
                    )
                    if office_mime is not None:
                        mime = office_mime

        if mime is not None and mime.lower() == "application/gzip":
            if detected_mime := _DocumentConversionInput._detect_mets_gbs(obj):
                mime = detected_mime

        mime = mime or _DocumentConversionInput._detect_html_xhtml(content)
        mime = mime or _DocumentConversionInput._detect_csv(content)
        mime = mime or "text/plain"
        formats = MimeTypeToFormat.get(mime, [])
        _log.info(f"detected formats: {formats}")

        if formats:
            if len(formats) == 1 and mime not in ("text/plain"):
                return formats[0]
            else:  # ambiguity in formats
                return _DocumentConversionInput._guess_from_content(
                    content, mime, formats, ext=obj_ext
                )
        else:
            return None

    @staticmethod
    def _has_doclang_extension(name: str) -> bool:
        lower_name = name.lower()
        return lower_name.endswith((".dclg", ".dclg.xml"))

    @staticmethod
    def _detect_office_mime_from_zip(
        source: Union[Path, BytesIO],
    ) -> Optional[str]:
        """Detect Office Open XML format by inspecting ZIP archive contents.

        Useful when the filename has no extension (e.g. pre-signed URLs)
        and filetype only reports ``application/zip``.
        """
        try:
            with zipfile.ZipFile(source) as zf:
                names = set(zf.namelist())
                mime_root = "application/vnd.openxmlformats-officedocument"
                if "word/document.xml" in names:
                    return mime_root + ".wordprocessingml.document"
                elif "xl/workbook.xml" in names:
                    return mime_root + ".spreadsheetml.sheet"
                elif "ppt/presentation.xml" in names:
                    return mime_root + ".presentationml.presentation"
                if "mimetype" in names:
                    odf_mime = zf.read("mimetype").decode("ascii", errors="ignore")
                    if odf_mime.startswith("application/vnd.oasis.opendocument."):
                        return odf_mime.strip()
        except (zipfile.BadZipFile, OSError):
            pass
        finally:
            if isinstance(source, BytesIO):
                source.seek(0)
        return None

    @staticmethod
    def _guess_from_content(
        content: bytes,
        mime: str,
        formats: list[InputFormat],
        ext: Optional[str] = None,
    ) -> Optional[InputFormat]:
        """Guess the input format of a document by checking part of its content."""
        input_format: Optional[InputFormat] = None

        if mime in {"application/xml", "application/xhtml+xml"}:
            content_str = content.decode("utf-8")

            if (
                InputFormat.XML_XBRL in formats
                and "http://www.xbrl.org/2003/instance" in content_str
                and "<xbrl" in content_str.lower()
            ):
                return InputFormat.XML_XBRL

            match_doctype = re.search(r"<!DOCTYPE [^>]+>", content_str)
            if match_doctype:
                xml_doctype = match_doctype.group()
                if InputFormat.XML_USPTO in formats and any(
                    item in xml_doctype
                    for item in (
                        "us-patent-application-v4",
                        "us-patent-grant-v4",
                        "us-grant-025",
                        "patent-application-publication",
                    )
                ):
                    input_format = InputFormat.XML_USPTO

                elif InputFormat.XML_JATS in formats and (
                    "JATS-journalpublishing" in xml_doctype
                    or "JATS-archive" in xml_doctype
                ):
                    input_format = InputFormat.XML_JATS

        elif mime == "text/plain":
            content_str = content.decode("utf-8", errors="replace")
            if InputFormat.XML_USPTO in formats and content_str.startswith("PATN\r\n"):
                input_format = InputFormat.XML_USPTO
            elif (
                InputFormat.MD in formats
                and ext is not None
                and ext.lower() in FormatToExtensions[InputFormat.MD]
            ):
                # Only fall back to MD when the extension is a known plain-text
                # extension (md/txt/text/qmd/rmd). Unknown extensions (e.g. .xyz)
                # must not be silently treated as Markdown.
                input_format = InputFormat.MD

        return input_format

    @staticmethod
    def _mime_from_extension(ext):
        mime = None
        if ext in FormatToExtensions[InputFormat.ASCIIDOC]:
            mime = FormatToMimeType[InputFormat.ASCIIDOC][0]
        elif ext in FormatToExtensions[InputFormat.HTML]:
            mime = FormatToMimeType[InputFormat.HTML][0]
        elif (
            ext in FormatToExtensions[InputFormat.XML_USPTO]
            and ext in FormatToExtensions[InputFormat.MD]
        ):
            # "txt" appears in both XML_USPTO and MD extension lists.  Leave mime=None
            # so the content-probing chain (_detect_html_xhtml, _detect_csv, then the
            # "text/plain" fallback + _guess_from_content) can pick the right format.
            pass
        elif ext in FormatToExtensions[InputFormat.MD]:
            mime = FormatToMimeType[InputFormat.MD][0]
        elif ext in FormatToExtensions[InputFormat.CSV]:
            mime = FormatToMimeType[InputFormat.CSV][0]
        elif ext in FormatToExtensions[InputFormat.JSON_DOCLING]:
            mime = FormatToMimeType[InputFormat.JSON_DOCLING][0]
        elif ext in FormatToExtensions[InputFormat.PDF]:
            mime = FormatToMimeType[InputFormat.PDF][0]
        elif ext in FormatToExtensions[InputFormat.DOCX]:
            mime = FormatToMimeType[InputFormat.DOCX][0]
        elif ext in FormatToExtensions[InputFormat.PPTX]:
            mime = FormatToMimeType[InputFormat.PPTX][0]
        elif ext in FormatToExtensions[InputFormat.XLSX]:
            mime = FormatToMimeType[InputFormat.XLSX][0]
        elif ext in FormatToExtensions[InputFormat.ODT]:
            mime = FormatToMimeType[InputFormat.ODT][0]
        elif ext in FormatToExtensions[InputFormat.ODS]:
            mime = FormatToMimeType[InputFormat.ODS][0]
        elif ext in FormatToExtensions[InputFormat.ODP]:
            mime = FormatToMimeType[InputFormat.ODP][0]
        elif ext in FormatToExtensions[InputFormat.VTT]:
            mime = FormatToMimeType[InputFormat.VTT][0]
        elif ext in FormatToExtensions[InputFormat.LATEX]:
            mime = FormatToMimeType[InputFormat.LATEX][0]
        elif ext in FormatToExtensions[InputFormat.EMAIL]:
            mime = FormatToMimeType[InputFormat.EMAIL][0]
        return mime

    @staticmethod
    def _detect_html_xhtml(
        content: bytes,
    ) -> Optional[Literal["application/xhtml+xml", "application/xml", "text/html"]]:
        """Guess the mime type of an XHTML, HTML, or XML file from its content.

        Args:
            content: A short piece of a document from its beginning.

        Returns:
            The mime type of an XHTML, HTML, or XML file, or None if the content does
              not match any of these formats.
        """
        content_str = content.decode("ascii", errors="ignore").lower()
        # Remove XML comments
        content_str = re.sub(r"<!--(.*?)-->", "", content_str, flags=re.DOTALL)
        content_str = content_str.lstrip()

        if re.match(r"<\?xml", content_str):
            if "xhtml" in content_str[:1000]:
                return "application/xhtml+xml"
            else:
                return "application/xml"

        if re.match(
            r"(<script.*?>.*?</script>\s*)?(<!doctype\s+html|<html|<head|<body)",
            content_str,
            re.DOTALL,
        ):
            return "text/html"

        p = re.compile(
            r"<!doctype\s+(?P<root>[a-zA-Z_:][a-zA-Z0-9_:.-]*)\s+.*>\s*<(?P=root)\b"
        )
        if p.search(content_str):
            return "application/xml"

        return None

    @staticmethod
    def _detect_csv(
        content: bytes,
    ) -> Optional[Literal["text/csv"]]:
        """Guess the mime type of a CSV file from its content.

        Args:
            content: A short piece of a document from its beginning.

        Returns:
            The mime type of a CSV file, or None if the content does
              not match any of the format.
        """
        content_str = content.decode("ascii", errors="ignore").strip()

        # Ensure there's at least one newline (CSV is usually multi-line)
        if "\n" not in content_str:
            return None

        # Use csv.Sniffer to detect CSV characteristics
        try:
            dialect = csv.Sniffer().sniff(content_str)
            if dialect.delimiter in {",", ";", "\t", "|"}:  # Common delimiters
                return "text/csv"
        except csv.Error:
            return None

        return None

    @staticmethod
    def _detect_mets_gbs(
        obj: Union[Path, DocumentStream],
    ) -> Optional[Literal["application/mets+xml"]]:
        # Use default limits for safe format detection
        default_options = MetsGbsBackendOptions()
        max_file_bytes = default_options.max_file_bytes
        max_member_count = default_options.max_member_count

        content = obj if isinstance(obj, Path) else obj.stream
        tar: tarfile.TarFile
        member: tarfile.TarInfo
        member_count = 0

        try:
            with tarfile.open(
                name=content if isinstance(content, Path) else None,
                fileobj=content if isinstance(content, BytesIO) else None,
                mode="r:gz",
            ) as tar:
                for member in tar.getmembers():
                    member_count += 1
                    if member_count > max_member_count:
                        _log.warning(
                            f"Archive exceeds member count limit ({max_member_count}) during format detection"
                        )
                        return None

                    if member.name.endswith(".xml"):
                        file = tar.extractfile(member)
                        if file is not None:
                            xml_content = file.read(max_file_bytes + 1)
                            if len(xml_content) > max_file_bytes:
                                _log.warning(
                                    f"XML file {member.name} exceeds size limit ({max_file_bytes} bytes) during format detection"
                                )
                                continue

                            content_str = xml_content.decode(errors="ignore")
                            if "http://www.loc.gov/METS/" in content_str:
                                return "application/mets+xml"
        except Exception as e:
            _log.warning(f"Error during METS-GBS format detection: {e}")
            return None

        return None
