From 31a7c5ed36a1cb72cc1d5305033189db71e37b10 Mon Sep 17 00:00:00 2001 From: Sai Teja Bandaru Date: Sat, 11 Jul 2026 01:47:13 +0200 Subject: [PATCH] feat(cli): add --info flag to extract lightweight JSON metadata without full conversion --- .../markitdown/src/markitdown/__main__.py | 18 ++ .../markitdown/src/markitdown/_markitdown.py | 184 ++++++++++++++++++ packages/markitdown/tests/test_cli_misc.py | 22 ++- packages/markitdown/tests/test_module_misc.py | 11 ++ 4 files changed, 233 insertions(+), 2 deletions(-) diff --git a/packages/markitdown/src/markitdown/__main__.py b/packages/markitdown/src/markitdown/__main__.py index ccb44b64b..bf47f1769 100644 --- a/packages/markitdown/src/markitdown/__main__.py +++ b/packages/markitdown/src/markitdown/__main__.py @@ -138,6 +138,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="Print a structured JSON summary of a file's metadata and exit.", + ) + parser.add_argument("filename", nargs="?") args = parser.parse_args() @@ -244,6 +250,18 @@ def main(): else: markitdown = MarkItDown(enable_plugins=args.use_plugins) + if args.info: + import json + try: + if args.filename is None: + metadata = markitdown.get_metadata(sys.stdin.buffer, stream_info=stream_info) + else: + metadata = markitdown.get_metadata(args.filename, stream_info=stream_info) + print(json.dumps(metadata, indent=2)) + sys.exit(0) + except Exception as e: + _exit_with_error(f"Error extracting metadata: {e}") + if args.filename is None: result = markitdown.convert_stream( sys.stdin.buffer, diff --git a/packages/markitdown/src/markitdown/_markitdown.py b/packages/markitdown/src/markitdown/_markitdown.py index f6aa4df0e..8aff690df 100644 --- a/packages/markitdown/src/markitdown/_markitdown.py +++ b/packages/markitdown/src/markitdown/_markitdown.py @@ -322,6 +322,190 @@ def convert( f"Invalid source type: {type(source)}. Expected str, requests.Response, BinaryIO." ) + def get_metadata( + self, + source: Union[str, requests.Response, Path, BinaryIO], + *, + stream_info: Optional[StreamInfo] = None, + **kwargs: Any, + ) -> Dict[str, Any]: + """ + Extract lightweight metadata for a file or stream (e.g. mimetype, detected converter) + without executing the full conversion pipeline. + """ + # Local path or URL + if isinstance(source, str): + if ( + source.startswith("http:") + or source.startswith("https:") + or source.startswith("file:") + or source.startswith("data:") + ): + # For URIs/URLs, convert them to a stream or resolve them + # Here we parse them or let requests handle them if it's a URL + # For simplicity, resolve via requests if it's an HTTP URL, otherwise treat as stream + if source.startswith("http:") or source.startswith("https:"): + response = requests.get(source, stream=True) + return self.get_metadata(response, stream_info=stream_info, **kwargs) + else: + # Parse local file URI + from ._uri_utils import parse_data_uri, file_uri_to_path + if source.startswith("data:"): + mime, charset, data = parse_data_uri(source) + stream = io.BytesIO(data) + res_stream_info = StreamInfo(mimetype=mime, charset=charset) + if stream_info: + res_stream_info = res_stream_info.copy_and_update(stream_info) + return self.get_metadata_stream(stream, stream_info=res_stream_info, **kwargs) + else: + netloc, path = file_uri_to_path(source) + return self.get_metadata_local(path, stream_info=stream_info, **kwargs) + else: + # Local path + return self.get_metadata_local(source, stream_info=stream_info, **kwargs) + + # Path object + if isinstance(source, Path): + return self.get_metadata_local(source, stream_info=stream_info, **kwargs) + + # Binary stream + if ( + hasattr(source, "read") + and callable(source.read) + and not isinstance(source, io.TextIOBase) + ): + return self.get_metadata_stream(source, stream_info=stream_info, **kwargs) + + # Requests Response + if isinstance(source, requests.Response): + stream = io.BytesIO(source.content) + url = source.url + filename = os.path.basename(urlparse(url).path) + content_type = source.headers.get("content-type") + mimetype = None + charset = None + if content_type: + parts = [p.strip() for p in content_type.split(";")] + mimetype = parts[0] + for p in parts[1:]: + if p.startswith("charset="): + charset = p[len("charset="):] + + res_stream_info = StreamInfo( + mimetype=mimetype, + charset=charset, + filename=filename, + url=url, + ) + if stream_info: + res_stream_info = res_stream_info.copy_and_update(stream_info) + return self.get_metadata_stream(stream, stream_info=res_stream_info, **kwargs) + + raise TypeError( + f"Invalid source type: {type(source)}. Expected str, requests.Response, BinaryIO, Path." + ) + + def get_metadata_local( + self, + path: Union[str, Path], + *, + stream_info: Optional[StreamInfo] = None, + **kwargs: Any, + ) -> Dict[str, Any]: + 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) + + size_bytes = os.path.getsize(path) + with open(path, "rb") as fh: + metadata = self.get_metadata_stream(fh, stream_info=base_guess, **kwargs) + metadata["path"] = path + metadata["size_bytes"] = size_bytes + return metadata + + def get_metadata_stream( + self, + stream: BinaryIO, + *, + stream_info: Optional[StreamInfo] = None, + **kwargs: Any, + ) -> Dict[str, Any]: + # Remember the initial stream position + cur_pos = stream.tell() + + # Calculate size by seeking to the end + try: + stream.seek(0, io.SEEK_END) + size_bytes = stream.tell() - cur_pos + except Exception: + size_bytes = None + finally: + stream.seek(cur_pos) + + base_guess = stream_info or StreamInfo() + guesses = self._get_stream_info_guesses(file_stream=stream, base_guess=base_guess) + + detected_converter = None + matched_mimetype = None + + sorted_registrations = sorted(self._converters, key=lambda x: x.priority) + + # Setup kwargs similar to _convert + _kwargs = {k: v for k, v in kwargs.items()} + 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 + _kwargs["_parent_converters"] = self._converters + + for guess_info in guesses + [StreamInfo()]: + for converter_registration in sorted_registrations: + converter = converter_registration.converter + + # Check if the converter accepts the stream + _accepts = False + try: + _kwargs_copy = _kwargs.copy() + if guess_info.extension is not None: + _kwargs_copy["file_extension"] = guess_info.extension + if guess_info.url is not None: + _kwargs_copy["url"] = guess_info.url + + _accepts = converter.accepts(stream, guess_info, **_kwargs_copy) + except Exception: + pass + finally: + stream.seek(cur_pos) + + if _accepts: + detected_converter = type(converter).__name__ + matched_mimetype = guess_info.mimetype + break + if detected_converter: + break + + metadata = { + "mime_type": matched_mimetype or base_guess.mimetype, + "detected_converter": detected_converter, + } + if size_bytes is not None: + metadata["size_bytes"] = size_bytes + + return metadata + def convert_local( self, path: Union[str, Path], diff --git a/packages/markitdown/tests/test_cli_misc.py b/packages/markitdown/tests/test_cli_misc.py index cf6c9ccc7..8ee8662c0 100644 --- a/packages/markitdown/tests/test_cli_misc.py +++ b/packages/markitdown/tests/test_cli_misc.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 -m pytest import subprocess +import sys from markitdown import __version__ # This file contains CLI tests that are not directly tested by the FileTestVectors. @@ -8,7 +9,7 @@ def test_version() -> None: result = subprocess.run( - ["python", "-m", "markitdown", "--version"], capture_output=True, text=True + [sys.executable, "-m", "markitdown", "--version"], capture_output=True, text=True ) assert result.returncode == 0, f"CLI exited with error: {result.stderr}" @@ -17,7 +18,7 @@ def test_version() -> None: def test_invalid_flag() -> None: result = subprocess.run( - ["python", "-m", "markitdown", "--foobar"], capture_output=True, text=True + [sys.executable, "-m", "markitdown", "--foobar"], capture_output=True, text=True ) assert result.returncode != 0, f"CLI exited with error: {result.stderr}" @@ -27,8 +28,25 @@ def test_invalid_flag() -> None: assert "SYNTAX" in result.stderr, "Expected 'SYNTAX' to appear in STDERR" +def test_info_flag() -> None: + import json + import os + tests_dir = os.path.dirname(__file__) + docx_file = os.path.join(tests_dir, "test_files", "test.docx") + result = subprocess.run( + [sys.executable, "-m", "markitdown", "--info", docx_file], capture_output=True, text=True + ) + assert result.returncode == 0, f"CLI exited with error: {result.stderr}" + data = json.loads(result.stdout) + assert data["detected_converter"] == "DocxConverter" + assert "application/vnd.openxmlformats-officedocument.wordprocessingml.document" in data["mime_type"] + assert data["size_bytes"] > 0 + assert data["path"] == docx_file + + if __name__ == "__main__": """Runs this file's tests from the command line.""" test_version() test_invalid_flag() + test_info_flag() print("All tests passed!") diff --git a/packages/markitdown/tests/test_module_misc.py b/packages/markitdown/tests/test_module_misc.py index 4d62e4919..5952db620 100644 --- a/packages/markitdown/tests/test_module_misc.py +++ b/packages/markitdown/tests/test_module_misc.py @@ -532,6 +532,16 @@ def test_markitdown_llm() -> None: validate_strings(result, PPTX_TEST_STRINGS) +def test_get_metadata() -> None: + markitdown = MarkItDown() + docx_file = os.path.join(TEST_FILES_DIR, "test.docx") + metadata = markitdown.get_metadata(docx_file) + assert metadata["detected_converter"] == "DocxConverter" + assert "application/vnd.openxmlformats-officedocument.wordprocessingml.document" in metadata["mime_type"] + assert metadata["size_bytes"] > 0 + assert metadata["path"] == docx_file + + if __name__ == "__main__": """Runs this file's tests from the command line.""" for test in [ @@ -547,6 +557,7 @@ def test_markitdown_llm() -> None: test_markitdown_exiftool, test_markitdown_llm_parameters, test_markitdown_llm, + test_get_metadata, ]: print(f"Running {test.__name__}...", end="") test()