Skills › Scientific Agent Skills › markitdown › batch_convert.py
batch_convert.py
354 lines · skills/markitdown/scripts/batch_convert.py at 5f2bc03
1#!/usr/bin/env python3
2"""Batch-convert trusted local files with Microsoft MarkItDown 0.1.8.
4The script deliberately uses convert_local(), skips symlinks, preserves relative
5directories, and keeps plugins disabled unless explicitly requested.
6"""
8from __future__ import annotations
10import argparse
11import json
12import sys
13from dataclasses import asdict, dataclass
14from datetime import datetime, timezone
15from importlib.metadata import version
16from pathlib import Path
17from tempfile import NamedTemporaryFile
18from typing import Iterable
20from markitdown import MarkItDown
23DEFAULT_EXTENSIONS = (
24 ".csv",
25 ".docx",
26 ".epub",
27 ".htm",
28 ".html",
29 ".ipynb",
30 ".jpeg",
31 ".jpg",
32 ".json",
33 ".msg",
34 ".pdf",
35 ".png",
36 ".pptx",
37 ".txt",
38 ".xls",
39 ".xlsx",
40 ".xml",
41)
43# With markitdown[all], these local files can trigger Google Web Speech.
44EXTERNAL_SERVICE_EXTENSIONS = {".m4a", ".mp3", ".mp4", ".wav"}
47@dataclass(slots=True)
48class ConversionRecord:
49 """One source file's conversion outcome."""
51 source: str
52 output: str | None
53 status: str
54 title: str | None = None
55 characters: int = 0
56 error: str | None = None
59def normalize_extensions(values: Iterable[str]) -> tuple[str, ...]:
60 """Normalize CLI extensions to unique lowercase values with leading dots."""
61 normalized: set[str] = set()
62 for value in values:
63 extension = value.strip().lower()
64 if not extension:
65 continue
66 if not extension.startswith("."):
67 extension = f".{extension}"
68 if extension == ".":
69 raise ValueError("A file extension cannot be only '.'")
70 normalized.add(extension)
71 if not normalized:
72 raise ValueError("At least one file extension is required")
73 return tuple(sorted(normalized))
76def discover_files(
77 input_dir: Path,
78 extensions: tuple[str, ...],
79 recursive: bool,
80) -> list[Path]:
81 """Return deterministic candidates without opening any files."""
82 iterator = input_dir.rglob("*") if recursive else input_dir.iterdir()
83 return sorted(
84 (
85 path
86 for path in iterator
87 if path.suffix.lower() in extensions
88 and (path.is_file() or path.is_symlink())
89 ),
90 key=lambda path: path.as_posix(),
91 )
94def output_path_for(source: Path, input_dir: Path, output_dir: Path) -> Path:
95 """Preserve directories and source suffix to avoid basename collisions."""
96 relative = source.relative_to(input_dir)
97 return output_dir / relative.parent / f"{relative.name}.md"
100def atomic_write_text(path: Path, content: str) -> None:
101 """Atomically replace a UTF-8 text file in its destination directory."""
102 path.parent.mkdir(parents=True, exist_ok=True)
103 temporary_path: Path | None = None
104 try:
105 with NamedTemporaryFile(
106 mode="w",
107 encoding="utf-8",
108 dir=path.parent,
109 prefix=f".{path.name}.",
110 suffix=".tmp",
111 delete=False,
112 ) as temporary:
113 temporary.write(content)
114 temporary_path = Path(temporary.name)
115 temporary_path.replace(path)
116 finally:
117 if temporary_path is not None and temporary_path.exists():
118 temporary_path.unlink()
121def convert_one(
122 converter: MarkItDown,
123 source: Path,
124 input_dir: Path,
125 output_dir: Path,
126 *,
127 overwrite: bool,
128 allow_empty: bool,
129 max_bytes: int,
130) -> ConversionRecord:
131 """Convert one local regular file and return a manifest record."""
132 relative_source = source.relative_to(input_dir).as_posix()
134 if source.is_symlink():
135 return ConversionRecord(
136 source=relative_source,
137 output=None,
138 status="skipped",
139 error="symbolic links are disabled",
140 )
142 try:
143 resolved = source.resolve(strict=True)
144 resolved.relative_to(input_dir)
145 if not resolved.is_file():
146 raise ValueError("source is not a regular file")
148 source_bytes = resolved.stat().st_size
149 if source_bytes > max_bytes:
150 return ConversionRecord(
151 source=relative_source,
152 output=None,
153 status="skipped",
154 error=f"source exceeds byte limit ({source_bytes} > {max_bytes})",
155 )
157 output = output_path_for(source, input_dir, output_dir)
158 relative_output = output.relative_to(output_dir).as_posix()
159 if output.exists() and not overwrite:
160 return ConversionRecord(
161 source=relative_source,
162 output=relative_output,
163 status="skipped",
164 error="output already exists",
165 )
167 result = converter.convert_local(resolved)
168 if not result.markdown.strip() and not allow_empty:
169 raise ValueError("conversion produced empty Markdown")
171 atomic_write_text(output, result.markdown)
172 return ConversionRecord(
173 source=relative_source,
174 output=relative_output,
175 status="converted",
176 title=result.title,
177 characters=len(result.markdown),
178 )
179 except Exception as exc: # Keep batch processing; record the exact failure.
180 return ConversionRecord(
181 source=relative_source,
182 output=None,
183 status="failed",
184 error=f"{type(exc).__name__}: {exc}",
185 )
188def write_manifest(
189 manifest_path: Path,
190 *,
191 input_dir: Path,
192 output_dir: Path,
193 records: list[ConversionRecord],
194) -> None:
195 """Write deterministic conversion metadata without source contents."""
196 payload = {
197 "generated_at": datetime.now(timezone.utc).isoformat(),
198 "markitdown_version": version("markitdown"),
199 "input_dir": str(input_dir),
200 "output_dir": str(output_dir),
201 "records": [asdict(record) for record in records],
202 }
203 atomic_write_text(
204 manifest_path,
205 json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
206 )
209def build_parser() -> argparse.ArgumentParser:
210 parser = argparse.ArgumentParser(
211 description=(
212 "Batch-convert trusted local files to Markdown. Output names retain "
213 "the source extension, for example paper.pdf.md."
214 )
215 )
216 parser.add_argument("input_dir", type=Path)
217 parser.add_argument("output_dir", type=Path)
218 parser.add_argument(
219 "--extensions",
220 nargs="+",
221 default=list(DEFAULT_EXTENSIONS),
222 metavar="EXT",
223 help="Extensions to include (default: common local document formats)",
224 )
225 parser.add_argument(
226 "--recursive",
227 "-r",
228 action="store_true",
229 help="Search subdirectories and preserve their relative paths",
230 )
231 parser.add_argument(
232 "--overwrite",
233 action="store_true",
234 help="Replace existing Markdown outputs",
235 )
236 parser.add_argument(
237 "--plugins",
238 action="store_true",
239 help="Enable installed plugins after reviewing and trusting them",
240 )
241 parser.add_argument(
242 "--allow-external-services",
243 action="store_true",
244 help=(
245 "Allow audio extensions that may invoke Google Web Speech. "
246 "Required for .wav, .mp3, .m4a, or .mp4."
247 ),
248 )
249 parser.add_argument(
250 "--allow-empty",
251 action="store_true",
252 help="Write empty Markdown instead of treating it as a failure",
253 )
254 parser.add_argument(
255 "--max-bytes",
256 type=int,
257 default=256 * 1024 * 1024,
258 help="Maximum source size in bytes (default: 268435456)",
259 )
260 parser.add_argument(
261 "--manifest",
262 type=Path,
263 help="Optional JSON manifest path",
264 )
265 parser.add_argument(
266 "--fail-fast",
267 action="store_true",
268 help="Stop after the first failed conversion",
269 )
270 return parser
273def main() -> int:
274 parser = build_parser()
275 args = parser.parse_args()
277 if args.max_bytes <= 0:
278 parser.error("--max-bytes must be positive")
280 try:
281 extensions = normalize_extensions(args.extensions)
282 except ValueError as exc:
283 parser.error(str(exc))
285 network_formats = sorted(set(extensions) & EXTERNAL_SERVICE_EXTENSIONS)
286 if network_formats and not args.allow_external_services:
287 parser.error(
288 "these formats can invoke an external transcription service: "
289 f"{', '.join(network_formats)}; pass --allow-external-services "
290 "only after user approval"
291 )
293 try:
294 input_dir = args.input_dir.resolve(strict=True)
295 except FileNotFoundError:
296 parser.error(f"input directory does not exist: {args.input_dir}")
297 if not input_dir.is_dir():
298 parser.error(f"input path is not a directory: {input_dir}")
300 candidates = discover_files(input_dir, extensions, args.recursive)
301 output_dir = args.output_dir.resolve()
302 output_dir.mkdir(parents=True, exist_ok=True)
304 if args.plugins:
305 print(
306 "WARNING: loading installed MarkItDown plugins into this process",
307 file=sys.stderr,
308 )
309 converter = MarkItDown(enable_plugins=args.plugins)
311 records: list[ConversionRecord] = []
312 for source in candidates:
313 record = convert_one(
314 converter,
315 source,
316 input_dir,
317 output_dir,
318 overwrite=args.overwrite,
319 allow_empty=args.allow_empty,
320 max_bytes=args.max_bytes,
321 )
322 records.append(record)
323 detail = record.output or record.error or ""
324 print(f"{record.status:9} {record.source} {detail}".rstrip())
325 if args.fail_fast and record.status == "failed":
326 break
328 if args.manifest is not None:
329 write_manifest(
330 args.manifest.resolve(),
331 input_dir=input_dir,
332 output_dir=output_dir,
333 records=records,
334 )
336 counts = {
337 status: sum(record.status == status for record in records)
338 for status in ("converted", "skipped", "failed")
339 }
340 print(
341 "Summary: "
342 f"{counts['converted']} converted, "
343 f"{counts['skipped']} skipped, "
344 f"{counts['failed']} failed"
345 )
347 if not candidates:
348 print(f"No matching files found for: {', '.join(extensions)}")
350 return 1 if counts["failed"] else 0
353if __name__ == "__main__":
354 raise SystemExit(main())