Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions packages/markitdown/src/markitdown/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from .__about__ import __version__
from ._markitdown import (
MarkItDown,
DocumentInfo,
PRIORITY_SPECIFIC_FILE_FORMAT,
PRIORITY_GENERIC_FILE_FORMAT,
)
Expand All @@ -21,6 +22,7 @@
__all__ = [
"__version__",
"MarkItDown",
"DocumentInfo",
"DocumentConverter",
"DocumentConverterResult",
"MarkItDownException",
Expand Down
31 changes: 31 additions & 0 deletions packages/markitdown/src/markitdown/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import argparse
import sys
import codecs
import json
from typing import Any, Dict
from textwrap import dedent
from importlib.metadata import entry_points
Expand Down Expand Up @@ -138,6 +139,12 @@ def main():
help="Keep data URIs (like base64-encoded images) in the output. By default, data URIs are truncated.",
)

parser.add_argument(
"--info",
action="store_true",
help="Emit lightweight document metadata as JSON without converting the file.",
)

parser.add_argument("filename", nargs="?")
args = parser.parse_args()

Expand Down Expand Up @@ -244,6 +251,26 @@ def main():
else:
markitdown = MarkItDown(enable_plugins=args.use_plugins)

if args.info:
if args.filename is None:
_exit_with_error("Filename is required when using --info.")
if _is_uri(args.filename):
_exit_with_error(
"--info currently supports local filesystem paths only (no URIs)."
)

try:
info = markitdown.get_local_document_info(
args.filename,
stream_info=stream_info,
keep_data_uris=args.keep_data_uris,
)
except (FileNotFoundError, OSError) as exc:
_exit_with_error(f"Could not read file: {exc}")

print(json.dumps(info.to_dict(), indent=2))
sys.exit(0)

if args.filename is None:
result = markitdown.convert_stream(
sys.stdin.buffer,
Expand Down Expand Up @@ -277,5 +304,9 @@ def _exit_with_error(message: str):
sys.exit(1)


def _is_uri(source: str) -> bool:
return source.startswith(("http:", "https:", "file:", "data:"))


if __name__ == "__main__":
main()
164 changes: 134 additions & 30 deletions packages/markitdown/src/markitdown/_markitdown.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import shutil
import traceback
import io
from dataclasses import dataclass
from dataclasses import asdict, dataclass
from importlib.metadata import entry_points
from typing import Any, List, Dict, Optional, Union, BinaryIO
from pathlib import Path
Expand Down Expand Up @@ -91,6 +91,26 @@ class ConverterRegistration:
priority: float


@dataclass(kw_only=True, frozen=True)
class DocumentInfo:
"""Lightweight metadata about a local document, without conversion."""

path: Optional[str]
size_bytes: Optional[int]
mime_type: Optional[str]
extension: Optional[str]
charset: Optional[str]
detected_converter: Optional[str]
page_count: Optional[int] = None
image_count: Optional[int] = None
table_count: Optional[int] = None
estimated_tokens: Optional[int] = None
warning: Optional[str] = None

def to_dict(self) -> Dict[str, Any]:
return asdict(self)


class MarkItDown:
"""(In preview) An extremely simple text-based document reader, suitable for LLM use.
This reader will convert common file-types or webpages to Markdown."""
Expand Down Expand Up @@ -359,6 +379,54 @@ def convert_local(
)
return self._convert(file_stream=fh, stream_info_guesses=guesses, **kwargs)

def get_local_document_info(
self,
path: Union[str, Path],
*,
stream_info: Optional[StreamInfo] = None,
**kwargs: Any,
) -> DocumentInfo:
"""Inspect a local document without running the conversion pipeline."""
if isinstance(path, Path):
path = str(path)

base_guess = StreamInfo(
local_path=path,
extension=os.path.splitext(path)[1],
filename=os.path.basename(path),
)

if stream_info is not None:
base_guess = base_guess.copy_and_update(stream_info)

with open(path, "rb") as fh:
guesses = self._get_stream_info_guesses(
file_stream=fh, base_guess=base_guess
)
detected_stream_info, detected_converter = self._detect_converter(
file_stream=fh, stream_info_guesses=guesses, **kwargs
)

info_stream = detected_stream_info
if info_stream is None and len(guesses) > 0:
info_stream = guesses[0]
if info_stream is None:
info_stream = base_guess

warning = None
if detected_converter is None:
warning = "No converter detected for this file."

return DocumentInfo(
path=path,
size_bytes=os.path.getsize(path),
mime_type=info_stream.mimetype,
extension=info_stream.extension,
charset=info_stream.charset,
detected_converter=detected_converter,
warning=warning,
)

def convert_stream(
self,
stream: BinaryIO,
Expand Down Expand Up @@ -582,34 +650,7 @@ def _convert(
cur_pos == file_stream.tell()
), "File stream position should NOT change between guess iterations"

_kwargs = {k: v for k, v in kwargs.items()}

# Copy any additional global options
if "llm_client" not in _kwargs and self._llm_client is not None:
_kwargs["llm_client"] = self._llm_client

if "llm_model" not in _kwargs and self._llm_model is not None:
_kwargs["llm_model"] = self._llm_model

if "llm_prompt" not in _kwargs and self._llm_prompt is not None:
_kwargs["llm_prompt"] = self._llm_prompt

if "style_map" not in _kwargs and self._style_map is not None:
_kwargs["style_map"] = self._style_map

if "exiftool_path" not in _kwargs and self._exiftool_path is not None:
_kwargs["exiftool_path"] = self._exiftool_path

# Add the list of converters for nested processing
_kwargs["_parent_converters"] = self._converters

# Add legaxy kwargs
if stream_info is not None:
if stream_info.extension is not None:
_kwargs["file_extension"] = stream_info.extension

if stream_info.url is not None:
_kwargs["url"] = stream_info.url
_kwargs = self._get_converter_kwargs(stream_info, kwargs)

# Check if the converter will accept the file, and if so, try to convert it
_accepts = False
Expand All @@ -621,7 +662,7 @@ def _convert(
# accept() should not have changed the file stream position
assert (
cur_pos == file_stream.tell()
), f"{type(converter).__name__}.accept() should NOT change the file_stream position"
), f"{type(converter).__name__}.accepts() should NOT change the file_stream position"

# Attempt the conversion
if _accepts:
Expand Down Expand Up @@ -653,6 +694,69 @@ def _convert(
"Could not convert stream to Markdown. No converter attempted a conversion, suggesting that the filetype is simply not supported."
)

def _detect_converter(
self, *, file_stream: BinaryIO, stream_info_guesses: List[StreamInfo], **kwargs
) -> tuple[Optional[StreamInfo], Optional[str]]:
sorted_registrations = sorted(self._converters, key=lambda x: x.priority)
cur_pos = file_stream.tell()

for stream_info in stream_info_guesses + [StreamInfo()]:
for converter_registration in sorted_registrations:
converter = converter_registration.converter
assert (
cur_pos == file_stream.tell()
), "File stream position should NOT change between guess iterations"

_kwargs = self._get_converter_kwargs(stream_info, kwargs)

_accepts = False
try:
_accepts = converter.accepts(file_stream, stream_info, **_kwargs)
except NotImplementedError:
pass

assert (
cur_pos == file_stream.tell()
), f"{type(converter).__name__}.accepts() should NOT change the file_stream position"

if _accepts:
return stream_info, type(converter).__name__

return None, None

def _get_converter_kwargs(
self, stream_info: StreamInfo, kwargs: Dict[str, Any]
) -> Dict[str, Any]:
_kwargs = {k: v for k, v in kwargs.items()}

# Copy any additional global options
if "llm_client" not in _kwargs and self._llm_client is not None:
_kwargs["llm_client"] = self._llm_client

if "llm_model" not in _kwargs and self._llm_model is not None:
_kwargs["llm_model"] = self._llm_model

if "llm_prompt" not in _kwargs and self._llm_prompt is not None:
_kwargs["llm_prompt"] = self._llm_prompt

if "style_map" not in _kwargs and self._style_map is not None:
_kwargs["style_map"] = self._style_map

if "exiftool_path" not in _kwargs and self._exiftool_path is not None:
_kwargs["exiftool_path"] = self._exiftool_path

# Add the list of converters for nested processing
_kwargs["_parent_converters"] = self._converters

# Add legacy kwargs
if stream_info.extension is not None:
_kwargs["file_extension"] = stream_info.extension

if stream_info.url is not None:
_kwargs["url"] = stream_info.url

return _kwargs

def register_page_converter(self, converter: DocumentConverter) -> None:
"""DEPRECATED: User register_converter instead."""
warn(
Expand Down
81 changes: 81 additions & 0 deletions packages/markitdown/tests/test_cli_misc.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,14 @@
#!/usr/bin/env python3 -m pytest
import json
import os
import subprocess
from markitdown import __version__

# This file contains CLI tests that are not directly tested by the FileTestVectors.
# This includes things like help messages, version numbers, and invalid flags.

TEST_FILES_DIR = os.path.join(os.path.dirname(__file__), "test_files")


def test_version() -> None:
result = subprocess.run(
Expand All @@ -27,8 +31,85 @@ def test_invalid_flag() -> None:
assert "SYNTAX" in result.stderr, "Expected 'SYNTAX' to appear in STDERR"


def test_info_outputs_xlsx_metadata_without_conversion() -> None:
test_file = os.path.join(TEST_FILES_DIR, "test.xlsx")

result = subprocess.run(
["python", "-m", "markitdown", "--info", test_file],
capture_output=True,
text=True,
)

assert result.returncode == 0, f"CLI exited with error: {result.stderr}"
info = json.loads(result.stdout)

assert set(info.keys()) == {
"path",
"size_bytes",
"mime_type",
"extension",
"charset",
"detected_converter",
"page_count",
"image_count",
"table_count",
"estimated_tokens",
"warning",
}
assert info["path"] == test_file
assert info["size_bytes"] == os.path.getsize(test_file)
assert info["mime_type"] is not None
assert info["extension"] == ".xlsx"
assert info["detected_converter"] == "XlsxConverter"
assert info["page_count"] is None
assert info["image_count"] is None
assert info["table_count"] is None
assert info["estimated_tokens"] is None
assert info["warning"] is None
assert "## Sheet1" not in result.stdout
assert "Alpha" not in result.stdout


def test_info_outputs_warning_for_unsupported_file() -> None:
test_file = os.path.join(TEST_FILES_DIR, "random.bin")

result = subprocess.run(
["python", "-m", "markitdown", "--info", test_file],
capture_output=True,
text=True,
)

assert result.returncode == 0, f"CLI exited with error: {result.stderr}"
info = json.loads(result.stdout)

assert info["path"] == test_file
assert info["size_bytes"] == os.path.getsize(test_file)
assert info["extension"] == ".bin"
assert info["detected_converter"] is None
assert info["warning"] == "No converter detected for this file."


def test_info_missing_local_file_outputs_clean_error() -> None:
missing_file = os.path.join(TEST_FILES_DIR, "missing-local-file.xlsx")

result = subprocess.run(
["python", "-m", "markitdown", "--info", missing_file],
capture_output=True,
text=True,
)

assert result.returncode != 0
assert "Could not read file:" in result.stdout
assert missing_file in result.stdout
assert "Traceback" not in result.stderr
assert "Traceback" not in result.stdout


if __name__ == "__main__":
"""Runs this file's tests from the command line."""
test_version()
test_invalid_flag()
test_info_outputs_xlsx_metadata_without_conversion()
test_info_outputs_warning_for_unsupported_file()
test_info_missing_local_file_outputs_clean_error()
print("All tests passed!")