Skills › Scientific Agent Skills › markitdown › convert_literature.py
convert_literature.py
405 lines · skills/markitdown/scripts/convert_literature.py at 5f2bc03
1#!/usr/bin/env python3
2"""Convert a trusted local PDF collection into provenance-rich Markdown."""
4from __future__ import annotations
6import argparse
7import json
8import re
9from dataclasses import asdict, dataclass
10from datetime import datetime, timezone
11from hashlib import sha256
12from importlib.metadata import version
13from pathlib import Path
14from tempfile import NamedTemporaryFile
15from urllib.parse import quote
17from markitdown import MarkItDown
20FILENAME_PATTERN = re.compile(
21 r"^(?P<author>.+?)_(?P<year>(?:19|20)\d{2})_(?P<title>.+)$"
22)
25@dataclass(slots=True)
26class LiteratureRecord:
27 """Conversion and inferred bibliography metadata for one PDF."""
29 source: str
30 output: str | None
31 status: str
32 title: str
33 author: str | None
34 year: str | None
35 source_sha256: str | None = None
36 characters: int = 0
37 error: str | None = None
40def humanize_filename_component(value: str) -> str:
41 """Turn underscore-delimited filename text into readable text."""
42 return re.sub(r"\s+", " ", value.replace("_", " ")).strip()
45def infer_metadata(path: Path) -> tuple[str | None, str | None, str]:
46 """Infer author/year/title from Author_Year_Title.pdf when possible."""
47 match = FILENAME_PATTERN.fullmatch(path.stem)
48 if match is None:
49 return None, None, humanize_filename_component(path.stem)
50 return (
51 humanize_filename_component(match.group("author")),
52 match.group("year"),
53 humanize_filename_component(match.group("title")),
54 )
57def digest_file(path: Path) -> str:
58 """Calculate SHA-256 without loading the complete PDF into memory."""
59 digest = sha256()
60 with path.open("rb") as stream:
61 for chunk in iter(lambda: stream.read(1024 * 1024), b""):
62 digest.update(chunk)
63 return digest.hexdigest()
66def yaml_scalar(value: str) -> str:
67 """JSON strings are valid, safely quoted YAML scalars."""
68 return json.dumps(value, ensure_ascii=False)
71def atomic_write_text(path: Path, content: str) -> None:
72 """Atomically replace a UTF-8 text file."""
73 path.parent.mkdir(parents=True, exist_ok=True)
74 temporary_path: Path | None = None
75 try:
76 with NamedTemporaryFile(
77 mode="w",
78 encoding="utf-8",
79 dir=path.parent,
80 prefix=f".{path.name}.",
81 suffix=".tmp",
82 delete=False,
83 ) as temporary:
84 temporary.write(content)
85 temporary_path = Path(temporary.name)
86 temporary_path.replace(path)
87 finally:
88 if temporary_path is not None and temporary_path.exists():
89 temporary_path.unlink()
92def output_path_for(
93 source: Path,
94 input_dir: Path,
95 output_dir: Path,
96 year: str | None,
97 organize_by_year: bool,
98) -> Path:
99 """Preserve source directories, optionally below a year directory."""
100 relative = source.relative_to(input_dir).with_suffix(".md")
101 if organize_by_year:
102 return output_dir / (year or "unknown-year") / relative
103 return output_dir / relative
106def render_document(
107 markdown: str,
108 *,
109 title: str,
110 author: str | None,
111 year: str | None,
112 source: str,
113 source_sha256: str,
114 converted_at: str,
115 markitdown_version: str,
116) -> str:
117 """Add minimal, safely quoted provenance front matter."""
118 fields = {
119 "title": title,
120 "author": author,
121 "year": year,
122 "source": source,
123 "source_sha256": source_sha256,
124 "converted_at": converted_at,
125 "markitdown_version": markitdown_version,
126 }
127 front_matter = ["---"]
128 for key, value in fields.items():
129 if value is not None:
130 front_matter.append(f"{key}: {yaml_scalar(value)}")
131 front_matter.extend(["---", ""])
132 return "\n".join(front_matter) + markdown
135def convert_paper(
136 converter: MarkItDown,
137 source: Path,
138 input_dir: Path,
139 output_dir: Path,
140 *,
141 organize_by_year: bool,
142 overwrite: bool,
143 allow_empty: bool,
144 max_bytes: int,
145 markitdown_version: str,
146) -> LiteratureRecord:
147 """Convert one local PDF."""
148 relative_source = source.relative_to(input_dir).as_posix()
149 author, year, inferred_title = infer_metadata(source)
150 output = output_path_for(
151 source,
152 input_dir,
153 output_dir,
154 year,
155 organize_by_year,
156 )
157 relative_output = output.relative_to(output_dir).as_posix()
159 if source.is_symlink():
160 return LiteratureRecord(
161 source=relative_source,
162 output=None,
163 status="skipped",
164 title=inferred_title,
165 author=author,
166 year=year,
167 error="symbolic links are disabled",
168 )
170 try:
171 resolved = source.resolve(strict=True)
172 resolved.relative_to(input_dir)
173 if not resolved.is_file():
174 raise ValueError("source is not a regular file")
176 source_bytes = resolved.stat().st_size
177 if source_bytes > max_bytes:
178 return LiteratureRecord(
179 source=relative_source,
180 output=None,
181 status="skipped",
182 title=inferred_title,
183 author=author,
184 year=year,
185 error=f"source exceeds byte limit ({source_bytes} > {max_bytes})",
186 )
188 if output.exists() and not overwrite:
189 return LiteratureRecord(
190 source=relative_source,
191 output=relative_output,
192 status="skipped",
193 title=inferred_title,
194 author=author,
195 year=year,
196 error="output already exists",
197 )
199 source_digest = digest_file(resolved)
200 result = converter.convert_local(resolved)
201 if not result.markdown.strip() and not allow_empty:
202 raise ValueError("conversion produced empty Markdown")
204 title = (result.title or "").strip() or inferred_title
205 converted_at = datetime.now(timezone.utc).isoformat()
206 document = render_document(
207 result.markdown,
208 title=title,
209 author=author,
210 year=year,
211 source=relative_source,
212 source_sha256=source_digest,
213 converted_at=converted_at,
214 markitdown_version=markitdown_version,
215 )
216 atomic_write_text(output, document)
217 return LiteratureRecord(
218 source=relative_source,
219 output=relative_output,
220 status="converted",
221 title=title,
222 author=author,
223 year=year,
224 source_sha256=source_digest,
225 characters=len(result.markdown),
226 )
227 except Exception as exc: # Continue the collection and record the failure.
228 return LiteratureRecord(
229 source=relative_source,
230 output=None,
231 status="failed",
232 title=inferred_title,
233 author=author,
234 year=year,
235 error=f"{type(exc).__name__}: {exc}",
236 )
239def escape_markdown(value: str) -> str:
240 """Escape link-label metacharacters."""
241 return value.replace("\\", "\\\\").replace("[", "\\[").replace("]", "\\]")
244def create_index(records: list[LiteratureRecord], output_dir: Path) -> None:
245 """Write a Markdown index and JSON catalog."""
246 eligible = [
247 record
248 for record in records
249 if record.output is not None and (output_dir / record.output).is_file()
250 ]
251 eligible.sort(
252 key=lambda record: (
253 record.year is None,
254 record.year or "",
255 record.title.casefold(),
256 record.source,
257 )
258 )
260 lines = [
261 "# Literature Index",
262 "",
263 f"Generated: {datetime.now(timezone.utc).isoformat()}",
264 f"Documents: {len(eligible)}",
265 "",
266 ]
267 grouped: dict[str, list[LiteratureRecord]] = {}
268 for record in eligible:
269 grouped.setdefault(record.year or "Unknown year", []).append(record)
271 for year in sorted(
272 grouped,
273 key=lambda value: (value == "Unknown year", value),
274 ):
275 lines.extend([f"## {year}", ""])
276 for record in grouped[year]:
277 label = escape_markdown(record.title)
278 link = quote(record.output or "", safe="/")
279 author = f" — {escape_markdown(record.author)}" if record.author else ""
280 lines.append(f"- [{label}]({link}){author}")
281 lines.append("")
283 atomic_write_text(output_dir / "INDEX.md", "\n".join(lines).rstrip() + "\n")
284 atomic_write_text(
285 output_dir / "catalog.json",
286 json.dumps(
287 [asdict(record) for record in records],
288 indent=2,
289 ensure_ascii=False,
290 )
291 + "\n",
292 )
295def build_parser() -> argparse.ArgumentParser:
296 parser = argparse.ArgumentParser(
297 description=(
298 "Convert trusted local literature PDFs to Markdown with provenance. "
299 "Filename convention: Author_Year_Title.pdf."
300 )
301 )
302 parser.add_argument("input_dir", type=Path)
303 parser.add_argument("output_dir", type=Path)
304 parser.add_argument(
305 "--recursive",
306 "-r",
307 action="store_true",
308 help="Search subdirectories and preserve their relative paths",
309 )
310 parser.add_argument(
311 "--organize-by-year",
312 action="store_true",
313 help="Place outputs below inferred year directories",
314 )
315 parser.add_argument(
316 "--create-index",
317 action="store_true",
318 help="Write INDEX.md and catalog.json",
319 )
320 parser.add_argument(
321 "--overwrite",
322 action="store_true",
323 help="Replace existing Markdown outputs",
324 )
325 parser.add_argument(
326 "--allow-empty",
327 action="store_true",
328 help="Write empty conversions instead of failing them",
329 )
330 parser.add_argument(
331 "--max-bytes",
332 type=int,
333 default=256 * 1024 * 1024,
334 help="Maximum PDF size in bytes (default: 268435456)",
335 )
336 return parser
339def main() -> int:
340 parser = build_parser()
341 args = parser.parse_args()
343 if args.max_bytes <= 0:
344 parser.error("--max-bytes must be positive")
346 try:
347 input_dir = args.input_dir.resolve(strict=True)
348 except FileNotFoundError:
349 parser.error(f"input directory does not exist: {args.input_dir}")
350 if not input_dir.is_dir():
351 parser.error(f"input path is not a directory: {input_dir}")
353 iterator = input_dir.rglob("*") if args.recursive else input_dir.iterdir()
354 papers = sorted(
355 (
356 path
357 for path in iterator
358 if path.suffix.lower() == ".pdf" and (path.is_file() or path.is_symlink())
359 ),
360 key=lambda path: path.as_posix(),
361 )
363 output_dir = args.output_dir.resolve()
364 output_dir.mkdir(parents=True, exist_ok=True)
365 package_version = version("markitdown")
366 converter = MarkItDown()
368 records: list[LiteratureRecord] = []
369 for paper in papers:
370 record = convert_paper(
371 converter,
372 paper,
373 input_dir,
374 output_dir,
375 organize_by_year=args.organize_by_year,
376 overwrite=args.overwrite,
377 allow_empty=args.allow_empty,
378 max_bytes=args.max_bytes,
379 markitdown_version=package_version,
380 )
381 records.append(record)
382 detail = record.output or record.error or ""
383 print(f"{record.status:9} {record.source} {detail}".rstrip())
385 if args.create_index:
386 create_index(records, output_dir)
388 counts = {
389 status: sum(record.status == status for record in records)
390 for status in ("converted", "skipped", "failed")
391 }
392 print(
393 "Summary: "
394 f"{counts['converted']} converted, "
395 f"{counts['skipped']} skipped, "
396 f"{counts['failed']} failed"
397 )
398 if not papers:
399 print("No PDF files found")
401 return 1 if counts["failed"] else 0
404if __name__ == "__main__":
405 raise SystemExit(main())