From 45de477adfd37bf4b24bac234d628c5e5e98c459 Mon Sep 17 00:00:00 2001 From: Justin Stuart Date: Tue, 10 Nov 2026 06:48:32 -0500 Subject: [PATCH] Add pdf_to_text script --- README.md | 2 + requirements.txt | 1 + scripts/pdf_to_txt.py | 89 +++++++++++++++++++++ tests/test_pdf_to_txt.py | 169 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 261 insertions(+) create mode 100644 scripts/pdf_to_txt.py create mode 100644 tests/test_pdf_to_txt.py diff --git a/README.md b/README.md index cbc0a47..09acce6 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ Scripts are fully tested, linted, and compatible with both direct execution and - **scripts/cbz_processor.py**: Compresses and optimizes CBZ files in a directory (see script for details). - **scripts/cbz_clean_up.py**: Recursively deletes CBZ backup files ending in `_original.cbz`. - **scripts/pdf_to_cbz.py**: Recursively converts PDF files into CBZ archives, then optimizes them with the CBZ processor. +- **scripts/pdf_to_txt.py**: Recursively extracts text from PDF files into plain `.txt` files alongside each source PDF. ## Requirements @@ -36,6 +37,7 @@ python -m scripts.mp3_metadata_stripper python -m scripts.cbz_processor python -m scripts.cbz_clean_up python -m scripts.pdf_to_cbz +python -m scripts.pdf_to_txt ``` ## Linting & Code Quality diff --git a/requirements.txt b/requirements.txt index fa35948..23e5225 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ Pillow pdf2image +pdfplumber coverage mutagen pylint diff --git a/scripts/pdf_to_txt.py b/scripts/pdf_to_txt.py new file mode 100644 index 0000000..59fdf52 --- /dev/null +++ b/scripts/pdf_to_txt.py @@ -0,0 +1,89 @@ +""" +PDF to TXT Utility + +Recursively scans a directory for PDF files, extracts text from each PDF, +and writes a plain text file alongside the source PDF. + +Features: + - Recursive processing of directories and subdirectories + - Layout-aware text extraction using pdfplumber + - Writes ".txt" next to each source PDF + - Pages are separated by a blank line in the output text + - Prints a summary of processed, skipped, and failed files + +Environment Variables: + PDF_TO_TXT_DIR (optional): Directory path to process files in +""" + +import os + +import pdfplumber +from dotenv import load_dotenv + +from scripts.utils import get_directory_from_env_or_prompt, make_summary_dict, print_result + +TITLES = { + "success": "Successfully converted", + "warning": "Skipped file", + "failed": "Failed to convert", +} + +def extract_page_text(page): + """Extract text from a single PDF page, preserving layout where possible.""" + try: + text = page.extract_text(layout=True) + except Exception: + text = page.extract_text() + if not text: + return "" + return "\n".join(line.rstrip() for line in text.splitlines()) + +def extract_text_from_pdf(pdf_path): + """Extract text from every page of a PDF, joined by blank lines.""" + pages = [] + with pdfplumber.open(pdf_path) as pdf: + for page in pdf.pages: + pages.append(extract_page_text(page) or "") + return "\n\n".join(pages) + +def convert_pdf_to_txt(pdf_path): + """Convert a PDF to a plain text file next to the source and return the path.""" + text = extract_text_from_pdf(pdf_path) + txt_path = f"{os.path.splitext(pdf_path)[0]}.txt" + with open(txt_path, "w", encoding="utf-8") as file: + file.write(text) + return txt_path + +def process_pdf_files(directory): + """Recursively process PDF files in a directory and return summary stats.""" + success_count = 0 + skipped_count = 0 + failed_count = 0 + + for root, _, files in os.walk(directory): + for filename in files: + file_path = os.path.join(root, filename) + if not filename.lower().endswith(".pdf"): + skipped_count += 1 + continue + + try: + convert_pdf_to_txt(file_path) + print(f"✅ Converted: {file_path}") + success_count += 1 + except Exception as error: + print(f"🛑 Failed to convert {file_path}: {error}") + failed_count += 1 + + return make_summary_dict(success_count, skipped_count, failed_count) + +def main(): + """Entry point for recursively converting PDFs to plain text files.""" + load_dotenv() + directory = get_directory_from_env_or_prompt("PDF_TO_TXT_DIR") + print(f"Processing directory: {directory}") + result = process_pdf_files(directory) + print_result(result, TITLES) + +if __name__ == "__main__": + main() diff --git a/tests/test_pdf_to_txt.py b/tests/test_pdf_to_txt.py new file mode 100644 index 0000000..763034f --- /dev/null +++ b/tests/test_pdf_to_txt.py @@ -0,0 +1,169 @@ +"""Tests for scripts/pdf_to_txt.py.""" + +from scripts import pdf_to_txt +from tests.test_helpers import run_cli_with_env + + +def make_pdf_bytes(text): + """Build a minimal single-page PDF containing the given text.""" + content_stream = f"BT /F1 24 Tf 72 720 Td ({text}) Tj ET".encode() + objects = [ + b"<< /Type /Catalog /Pages 2 0 R >>", + b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>", + ( + b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] " + b"/Contents 4 0 R /Resources << /Font << /F1 5 0 R >> >> >>" + ), + ( + b"<< /Length " + + str(len(content_stream)).encode() + + b" >>\nstream\n" + + content_stream + + b"\nendstream" + ), + b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>", + ] + + output = bytearray(b"%PDF-1.4\n") + offsets = [] + for index, obj in enumerate(objects, start=1): + offsets.append(len(output)) + output += f"{index} 0 obj\n".encode() + obj + b"\nendobj\n" + + xref_offset = len(output) + output += f"xref\n0 {len(objects) + 1}\n".encode() + output += b"0000000000 65535 f \n" + for offset in offsets: + output += f"{offset:010d} 00000 n \n".encode() + output += ( + f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\n" + f"startxref\n{xref_offset}\n%%EOF\n" + ).encode() + return bytes(output) + + +class DummyPage: + """Simple page test double that returns canned text.""" + + def __init__(self, text, layout_error=False): + self.text = text + self.layout_error = layout_error + + def extract_text(self, layout=False): + """Return canned text, raising for layout mode when configured.""" + if layout and self.layout_error: + raise ValueError("layout failed") + return self.text + + +class DummyPdf: + """Simple pdf test double exposing a pages list.""" + + def __init__(self, pages): + self.pages = pages + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + +def test_convert_pdf_to_txt_writes_text_next_to_pdf(monkeypatch, tmp_path): + """Conversion should write a .txt file next to the source PDF.""" + pdf_path = tmp_path / "story.pdf" + pdf_path.write_bytes(b"%PDF-1.4") + + monkeypatch.setattr( + pdf_to_txt, + "pdfplumber", + type("M", (), {"open": staticmethod(lambda _: DummyPdf([DummyPage("Hello"), DummyPage("World")]))}), + ) + + txt_path = pdf_to_txt.convert_pdf_to_txt(str(pdf_path)) + + assert txt_path == str(tmp_path / "story.txt") + assert (tmp_path / "story.txt").read_text(encoding="utf-8") == "Hello\n\nWorld" + assert pdf_path.exists() + + +def test_extract_text_from_pdf_falls_back_when_layout_fails(monkeypatch, tmp_path): + """Layout extraction failures should fall back to default extraction.""" + pdf_path = tmp_path / "story.pdf" + pdf_path.write_bytes(b"%PDF-1.4") + + monkeypatch.setattr( + pdf_to_txt, + "pdfplumber", + type( + "M", + (), + {"open": staticmethod(lambda _: DummyPdf([DummyPage("Fallback", layout_error=True)]))}, + ), + ) + + assert pdf_to_txt.extract_text_from_pdf(str(pdf_path)) == "Fallback" + + +def test_process_pdf_files_recurses_and_skips_non_pdf(monkeypatch, tmp_path): + """Only PDF files should be processed while others are counted as skipped.""" + pdf_file = tmp_path / "docs" / "book.pdf" + pdf_file.parent.mkdir() + text_file = tmp_path / "docs" / "index.txt" + text_file.write_text("skip") + pdf_file.write_bytes(b"%PDF-1.4") + + seen = [] + + def fake_convert(path): + seen.append(path) + return f"{path}.txt" + + monkeypatch.setattr(pdf_to_txt, "convert_pdf_to_txt", fake_convert) + + result = pdf_to_txt.process_pdf_files(str(tmp_path)) + + assert result["success_count"] == 1 + assert result["failed_count"] == 0 + assert result["skipped_count"] == 1 + assert seen == [str(pdf_file)] + + +def test_process_pdf_files_counts_failures(monkeypatch, tmp_path): + """Failed PDF conversions should increment failed_count and continue.""" + pdf_file = tmp_path / "broken.pdf" + pdf_file.write_bytes(b"%PDF-1.4") + + def fake_convert(*_args, **_kwargs): + raise RuntimeError("bad pdf") + + monkeypatch.setattr(pdf_to_txt, "convert_pdf_to_txt", fake_convert) + + result = pdf_to_txt.process_pdf_files(str(tmp_path)) + + assert result["failed_count"] == 1 + assert result["success_count"] == 0 + assert result["skipped_count"] == 0 + + +def test_end_to_end_with_real_pdf(tmp_path): + """A real minimal PDF should convert to a txt file containing its text.""" + pdf_file = tmp_path / "real.pdf" + pdf_file.write_bytes(make_pdf_bytes("Hello from PDF")) + + result = pdf_to_txt.process_pdf_files(str(tmp_path)) + + assert result["success_count"] == 1 + assert result["failed_count"] == 0 + output = (tmp_path / "real.txt").read_text(encoding="utf-8") + assert "Hello" in output + assert "PDF" in output + assert all(line == line.rstrip() for line in output.splitlines()) + + +def test_cli_entry(tmp_path): + """CLI module execution should run with PDF_TO_TXT_DIR set.""" + env_vars = {"PDF_TO_TXT_DIR": str(tmp_path)} + result = run_cli_with_env("scripts.pdf_to_txt", env_vars) + assert result is not None + assert result.returncode == 0