"""
cv_converter.py
===============
Converts CV / résumé files (PDF, DOCX, DOC) to clean Markdown.
Pipeline
--------
PDF → pdfplumber (scan detection) → Marker (with or without OCR)
DOCX → pypandoc → GitHub-Flavoured Markdown
DOC → LibreOffice (→ DOCX) → pypandoc → GFM
Why Markdown as the intermediate format?
• LLMs understand it natively → better job-match prompts
• Sentence-transformers get cleaner text → better embeddings
• Renders directly in the browser with zero extra work
Install
-------
pip install marker-pdf pdfplumber pypandoc
python -c "import pypandoc; pypandoc.download_pandoc()"
# For legacy .doc support:
# Ubuntu/Debian : sudo apt-get install libreoffice
# macOS : brew install --cask libreoffice
Usage
-----
from cv_converter import CVConverter
converter = CVConverter()
result = converter.convert("john_doe_cv.pdf")
if result: # bool(result) == result.success
print(result.markdown)
converter.save(result, "john_doe_cv.md")
else:
print("Failed:", result.error)
for w in result.warnings:
print("Warning:", w)
"""
from __future__ import annotations
import logging
import re
import subprocess
import tempfile
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Optional
logger = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# Value types
# ─────────────────────────────────────────────────────────────────────────────
class ConversionMethod(str, Enum):
MARKER = "marker" # Marker – text-based PDF
MARKER_OCR = "marker_ocr" # Marker – forced OCR (scanned PDF)
PANDOC = "pandoc" # Pandoc – DOCX
PANDOC_VIA_LO = "pandoc_via_lo" # LibreOffice → Pandoc – legacy DOC
class FileType(str, Enum):
PDF = "pdf"
DOCX = "docx"
DOC = "doc"
UNKNOWN = "unknown"
@dataclass
class ConversionResult:
"""
Returned by :py:meth:`CVConverter.convert`.
Always check ``.success`` (or ``bool(result)``) before reading
``.markdown``.
Attributes
----------
success : bool – True when conversion produced output.
markdown : str | None – The Markdown text (None on failure).
method_used : str | None – Which pipeline was used.
file_type : str | None – Detected file type ("pdf", "docx", …).
is_scanned : bool – True when OCR was required.
page_count : int – Page count (0 when unknown).
warnings : list[str] – Non-fatal notes (mixed PDF, OCR quality…).
error : str | None – Human-readable error message on failure.
"""
success: bool
markdown: Optional[str] = None
method_used: Optional[str] = None
file_type: Optional[str] = None
is_scanned: bool = False
page_count: int = 0
warnings: list[str] = field(default_factory=list)
error: Optional[str] = None
def __bool__(self) -> bool:
return self.success
# ─────────────────────────────────────────────────────────────────────────────
# Main class
# ─────────────────────────────────────────────────────────────────────────────
class CVConverter:
"""
Converts CV files (PDF / DOCX / DOC) to clean Markdown.
Marker ML models are loaded lazily on the first PDF call and then
cached, so all subsequent conversions in the same process reuse them.
Parameters
----------
temp_dir : str | None
Where to place intermediate files (e.g. the .docx produced when
converting a legacy .doc). Defaults to the OS temp directory.
marker_device : str
``"cpu"`` or ``"cuda"``. Passed to Marker when loading models.
Examples
--------
>>> converter = CVConverter()
>>> result = converter.convert("resume.pdf")
>>> if result:
... converter.save(result, "resume.md")
"""
# ── tuneable thresholds ───────────────────────────────────────────────
# Characters needed on a page before we call it "text bearing"
MIN_TEXT_CHARS_PER_PAGE: int = 50
# Fraction of pages that must carry text to skip OCR
SCANNED_PAGE_RATIO: float = 0.30
# Hard timeout for the LibreOffice subprocess (seconds)
LIBREOFFICE_TIMEOUT: int = 60
# ─────────────────────────────────────────────────────────────────────
def __init__(
self,
temp_dir: Optional[str] = None,
marker_device: str = "cpu",
) -> None:
self.temp_dir = Path(temp_dir or tempfile.gettempdir())
self.marker_device = marker_device
self._marker_models = None # populated on first use
# ─────────────────────────────────────────────────────────────────────
# Public API
# ─────────────────────────────────────────────────────────────────────
def convert(self, file_path: str | Path) -> ConversionResult:
"""
Convert a CV file to Markdown.
Automatically detects the file type and delegates to the correct
sub-pipeline. Never raises; all errors are captured in the
returned :class:`ConversionResult`.
Parameters
----------
file_path : str | Path
Path to the CV file (.pdf, .docx, or .doc).
Returns
-------
ConversionResult
"""
path = Path(file_path)
# ── pre-flight checks ─────────────────────────────────────────────
if not path.exists():
return ConversionResult(
success=False, error=f"File not found: {path}"
)
if not path.is_file():
return ConversionResult(
success=False, error=f"Path is not a regular file: {path}"
)
if path.stat().st_size == 0:
return ConversionResult(
success=False, error=f"File is empty: {path.name}"
)
file_type = self._detect_file_type(path)
if file_type == FileType.PDF:
return self._convert_pdf(path)
if file_type in (FileType.DOCX, FileType.DOC):
return self._convert_word(path, file_type)
return ConversionResult(
success=False,
error=(
f"Unsupported file type '{path.suffix}'. "
"Accepted: .pdf, .docx, .doc"
),
)
def save(
self,
result: ConversionResult,
output_path: str | Path,
) -> None:
"""
Write ``result.markdown`` to *output_path* (UTF-8).
Raises
------
ValueError
When ``result.success`` is False.
"""
if not result.success:
raise ValueError("Cannot save a failed ConversionResult.")
out = Path(output_path)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(result.markdown, encoding="utf-8")
logger.info("Markdown written → %s", out)
def convert_and_save(
self,
file_path: str | Path,
output_path: str | Path,
) -> ConversionResult:
"""Convenience wrapper: convert then save if successful."""
result = self.convert(file_path)
if result.success:
self.save(result, output_path)
return result
# ─────────────────────────────────────────────────────────────────────
# File-type detection
# ─────────────────────────────────────────────────────────────────────
@staticmethod
def _detect_file_type(path: Path) -> FileType:
return {
".pdf": FileType.PDF,
".docx": FileType.DOCX,
".doc": FileType.DOC,
}.get(path.suffix.lower(), FileType.UNKNOWN)
# ─────────────────────────────────────────────────────────────────────
# PDF pipeline
# ─────────────────────────────────────────────────────────────────────
def _check_pdf_text_layer(
self, path: Path
) -> tuple[bool, int, list[bool]]:
"""
Use pdfplumber to probe each page for an embedded text layer.
Returns
-------
(has_text_layer, page_count, per_page_flags)
* has_text_layer – True when enough pages carry real text.
* page_count – Total pages.
* per_page_flags – Per-page bool list (True = text found).
Raises
------
ValueError
For password-protected or zero-page PDFs.
"""
try:
import pdfplumber
except ImportError:
logger.warning(
"pdfplumber not installed — OCR detection skipped. "
"Run: pip install pdfplumber"
)
return True, 0, []
try:
with pdfplumber.open(str(path)) as pdf:
page_count = len(pdf.pages)
if page_count == 0:
raise ValueError(
f"'{path.name}' has zero pages and cannot be converted."
)
per_page: list[bool] = []
for page in pdf.pages:
raw_text = page.extract_text() or ""
per_page.append(
len(raw_text.strip()) >= self.MIN_TEXT_CHARS_PER_PAGE
)
pages_with_text = sum(per_page)
ratio = pages_with_text / page_count
has_text_layer = ratio >= self.SCANNED_PAGE_RATIO
logger.debug(
"Text-layer check: %d/%d pages have text (%.0f%%)",
pages_with_text, page_count, ratio * 100,
)
return has_text_layer, page_count, per_page
except Exception as exc:
msg = str(exc).lower()
if any(k in msg for k in ("password", "encrypted", "decrypt")):
raise ValueError(
f"'{path.name}' is password-protected. "
"Please provide an unlocked copy."
) from exc
# Unknown pdfplumber error → assume text-based; Marker will cope.
logger.warning(
"pdfplumber probe failed (%s) — assuming text-based PDF.", exc
)
return True, 0, []
def _load_marker_models(self):
"""Lazy-load and cache Marker model dict (once per process).
PyTorch lazy / meta-device initialization can cause:
'Cannot copy out of meta tensor; no data!'
when Marker calls model.to(device) on a model that was built on the
meta device. We patch torch.nn.Module.to to fall back to to_empty()
in that case, then restore the original after loading completes.
"""
if self._marker_models is None:
logger.info(
"Loading Marker models for the first time "
"(this may take ~10–30 s)…"
)
try:
import torch
_original_to = torch.nn.Module.to
def _safe_to(module, *args, **kwargs):
try:
return _original_to(module, *args, **kwargs)
except RuntimeError as exc:
if "Cannot copy out of meta tensor" in str(exc):
device = args[0] if args else kwargs.get("device", "cpu")
logger.debug(
"Meta-tensor detected — using to_empty(device=%s)", device
)
return module.to_empty(device=device)
raise
torch.nn.Module.to = _safe_to
try:
from marker.models import create_model_dict
self._marker_models = create_model_dict()
logger.info("Marker models loaded and cached.")
finally:
# Always restore the original .to even if loading fails
torch.nn.Module.to = _original_to
except ImportError as exc:
raise ImportError(
"Marker is not installed. Fix: pip install marker-pdf"
) from exc
return self._marker_models
def _run_marker(self, path: Path, force_ocr: bool = False) -> str:
"""
Execute Marker and return the raw Markdown string.
Parameters
----------
force_ocr : bool
When True, Marker is told to OCR every page regardless of
whether it detects an embedded text layer.
Raises
------
RuntimeError
When Marker raises or returns empty output.
"""
model_dict = self._load_marker_models()
try:
from marker.converters.pdf import PdfConverter # v1.x API
from marker.output import text_from_rendered # v1.x API
config = {"force_ocr": force_ocr} if force_ocr else {}
converter = PdfConverter(model_dict, config=config)
rendered = converter(str(path))
full_text, _images, _meta = text_from_rendered(rendered)
except Exception as exc:
raise RuntimeError(
f"Marker raised an exception on '{path.name}': {exc}"
) from exc
if not full_text or not full_text.strip():
raise RuntimeError(
f"Marker produced empty output for '{path.name}'. "
"The file may be image-only or severely corrupted."
)
return full_text
def _convert_pdf(self, path: Path) -> ConversionResult:
"""
Full PDF → Markdown pipeline.
Decision tree
-------------
1. pdfplumber inspects every page for a text layer.
2a. Fully scanned (< SCANNED_PAGE_RATIO text pages)
→ Marker + OCR.
2b. Mixed pages (some pages lack text)
→ Marker + OCR for consistency across all pages.
2c. Fully digital → Marker without OCR.
3. If step 2c fails, automatically retry with OCR as a fallback
(handles PDFs that report a text layer but are actually images).
"""
warnings: list[str] = []
# ── 1. detect text layer ─────────────────────────────────────────
try:
has_text, page_count, per_page = self._check_pdf_text_layer(path)
except ValueError as exc:
return ConversionResult(success=False, error=str(exc))
# ── 2. decide OCR strategy ───────────────────────────────────────
scanned_page_nums: list[int] = [
i + 1 for i, flag in enumerate(per_page) if not flag
]
is_mixed = bool(scanned_page_nums) and has_text
is_scanned = not has_text
if is_scanned:
force_ocr = True
method = ConversionMethod.MARKER_OCR
warnings.append(
"Document appears to be fully scanned. OCR was applied — "
"accuracy depends on scan quality."
)
elif is_mixed:
force_ocr = True
is_scanned = True
method = ConversionMethod.MARKER_OCR
warnings.append(
f"Mixed PDF: pages {scanned_page_nums} appear scanned. "
"OCR applied to the entire document for consistency."
)
else:
force_ocr = False
method = ConversionMethod.MARKER
# ── 3. convert ───────────────────────────────────────────────────
try:
markdown = self._run_marker(path, force_ocr=force_ocr)
except RuntimeError as exc:
if not force_ocr:
# Rare edge case: PDF claims a text layer but extraction
# fails (e.g. custom fonts, badly embedded text).
logger.warning(
"Marker (no OCR) failed on '%s': %s — retrying with OCR…",
path.name, exc,
)
try:
markdown = self._run_marker(path, force_ocr=True)
method = ConversionMethod.MARKER_OCR
is_scanned = True
warnings.append(
"Standard text extraction failed; OCR fallback was used."
)
except RuntimeError as fallback_exc:
return ConversionResult(
success=False,
error=(
f"All extraction methods failed for '{path.name}'. "
f"Last error: {fallback_exc}"
),
)
else:
return ConversionResult(success=False, error=str(exc))
return ConversionResult(
success = True,
markdown = self._clean_markdown(markdown),
method_used = method.value,
file_type = FileType.PDF.value,
is_scanned = is_scanned,
page_count = page_count,
warnings = warnings,
)
# ─────────────────────────────────────────────────────────────────────
# Word pipeline
# ─────────────────────────────────────────────────────────────────────
def _convert_word(
self, path: Path, file_type: FileType
) -> ConversionResult:
"""
Convert DOCX (or legacy DOC) to GitHub-Flavoured Markdown via Pandoc.
For ``.doc`` files LibreOffice is used first to produce a ``.docx``
intermediate in ``self.temp_dir``, which Pandoc then converts.
Pandoc flags used
-----------------
``--wrap=none`` No hard line-wrapping.
``--strip-comments`` Drop Word tracked-change comments.
``--markdown-headings=atx`` Use ``#`` style, not underline style.
"""
try:
import pypandoc # noqa: F401 – just check it is installed
except ImportError:
return ConversionResult(
success=False,
error=(
"pypandoc is not installed.\n"
" pip install pypandoc\n"
" python -c \"import pypandoc; pypandoc.download_pandoc()\""
),
)
warnings: list[str] = []
method = ConversionMethod.PANDOC
# ── .doc → .docx ─────────────────────────────────────────────────
if file_type == FileType.DOC:
try:
path, warnings = self._doc_to_docx(path, warnings)
method = ConversionMethod.PANDOC_VIA_LO
except RuntimeError as exc:
return ConversionResult(success=False, error=str(exc))
# ── DOCX → GFM Markdown ──────────────────────────────────────────
try:
import pypandoc
markdown = pypandoc.convert_file(
str(path),
to="gfm",
extra_args=[
"--wrap=none",
"--strip-comments",
"--markdown-headings=atx",
],
)
except Exception as exc:
return ConversionResult(
success=False,
error=f"Pandoc conversion failed: {exc}",
)
if not markdown or not markdown.strip():
return ConversionResult(
success=False,
error=(
f"Pandoc returned empty output for '{path.name}'. "
"The document may be blank."
),
)
return ConversionResult(
success = True,
markdown = self._clean_markdown(markdown),
method_used = method.value,
file_type = file_type.value,
is_scanned = False,
warnings = warnings,
)
def _doc_to_docx(
self, path: Path, warnings: list[str]
) -> tuple[Path, list[str]]:
"""
Convert a legacy ``.doc`` file to ``.docx`` via LibreOffice (headless).
Returns
-------
(docx_path, updated_warnings)
Raises
------
RuntimeError
If LibreOffice is missing, exits non-zero, times out, or
produces no output file.
"""
docx_path = self.temp_dir / (path.stem + ".docx")
try:
proc = subprocess.run(
[
"libreoffice",
"--headless",
"--convert-to", "docx",
"--outdir", str(self.temp_dir),
str(path),
],
capture_output=True,
text=True,
timeout=self.LIBREOFFICE_TIMEOUT,
)
except FileNotFoundError:
raise RuntimeError(
"LibreOffice is required to convert legacy .doc files but "
"was not found on PATH.\n"
" Ubuntu/Debian : sudo apt-get install libreoffice\n"
" macOS : brew install --cask libreoffice"
)
except subprocess.TimeoutExpired:
raise RuntimeError(
f"LibreOffice timed out after {self.LIBREOFFICE_TIMEOUT} s "
f"converting '{path.name}'."
)
if proc.returncode != 0:
raise RuntimeError(
f"LibreOffice exited with code {proc.returncode}:\n"
f"{proc.stderr.strip()}"
)
if not docx_path.exists():
raise RuntimeError(
f"LibreOffice ran successfully but no .docx was produced. "
f"Expected: {docx_path}"
)
warnings.append(
"Legacy .doc file was automatically converted to .docx via "
"LibreOffice before Markdown extraction."
)
return docx_path, warnings
# ─────────────────────────────────────────────────────────────────────
# Post-processing helpers
# ─────────────────────────────────────────────────────────────────────
@staticmethod
def _clean_markdown(text: str) -> str:
"""
Remove converter artefacts and normalise whitespace.
Handles
-------
* Windows / mixed line endings (CRLF → LF)
* Null bytes and Word private-use bullet characters
* Trailing whitespace per line
* Runs of 3+ blank lines → single blank line
* Marker page-break separators (standalone ``---`` / ``===`` lines)
* Pandoc footnote-like wrapping artefacts
"""
# 1. Normalise line endings
text = text.replace("\r\n", "\n").replace("\r", "\n")
# 2. Remove null bytes and common Word private-use characters
replacements = {
"\x00": "", # null byte
"\uf0b7": "-", # Word solid bullet •
"\uf0a7": "-", # Word hollow bullet ◦
"\uf020": " ", # Word private space
"\uf0fc": "-", # Word checkmark bullet
}
for bad, good in replacements.items():
text = text.replace(bad, good)
# 3. Strip trailing whitespace on every line
text = "\n".join(line.rstrip() for line in text.split("\n"))
# 4. Remove Marker's standalone page-break / rule lines
text = re.sub(r"(?m)^[-=]{3,}\s*$", "", text)
# 5. Collapse 3+ consecutive blank lines to one blank line
text = re.sub(r"\n{3,}", "\n\n", text)
return text.strip()
# ─────────────────────────────────────────────────────────────────────────────
# CLI convenience (python cv_converter.py [output])
# ─────────────────────────────────────────────────────────────────────────────
def _cli() -> None:
import sys
logging.basicConfig(
level=logging.INFO,
format="%(levelname)s %(message)s",
)
args = sys.argv[1:]
if not args:
print(
"Usage: python cv_converter.py [output.md]"
)
sys.exit(1)
input_path = Path(args[0])
output_path = Path(args[1]) if len(args) > 1 else input_path.with_suffix(".md")
converter = CVConverter()
print(f"Converting '{input_path}' …")
result = converter.convert(input_path)
if result.warnings:
for w in result.warnings:
print(f" ⚠ {w}")
if not result:
print(f"✗ Conversion failed: {result.error}")
sys.exit(1)
converter.save(result, output_path)
print(
f"✓ Done [{result.method_used}] "
f"{'(scanned) ' if result.is_scanned else ''}"
f"→ {output_path}"
)
if result.page_count:
print(f" Pages : {result.page_count}")
print(f" Size : {len(result.markdown):,} characters")
if __name__ == "__main__":
_cli()