diff --git a/crawl4ai/processors/pdf/processor.py b/crawl4ai/processors/pdf/processor.py index 96ff0d260..ebcd53df1 100644 --- a/crawl4ai/processors/pdf/processor.py +++ b/crawl4ai/processors/pdf/processor.py @@ -66,7 +66,6 @@ def __init__(self, image_dpi: int = 144, image_quality: int = 85, extract_images self.image_dpi = image_dpi self.image_quality = image_quality - self.current_page_number = 0 self.extract_images = extract_images self.save_images_locally = save_images_locally self.image_save_dir = image_save_dir @@ -122,8 +121,7 @@ def process(self, pdf_path: Path) -> PDFProcessResult: for page_num, page in enumerate(reader.pages): if page_num >= page_limit: break - self.current_page_number = page_num + 1 - pdf_page = self._process_page(page, image_dir) + pdf_page = self._process_page(page, image_dir, page_num + 1) result.pages.append(pdf_page) except Exception as e: @@ -186,8 +184,7 @@ def process_page_safely(page_num: int): with pdf_path.open('rb') as file: thread_reader = PdfReader(file) page = thread_reader.pages[page_num] - self.current_page_number = page_num + 1 - return self._process_page(page, image_dir) + return self._process_page(page, image_dir, page_num + 1) # Process pages in parallel batches with concurrent.futures.ThreadPoolExecutor(max_workers=self.batch_size) as executor: @@ -221,9 +218,9 @@ def process_page_safely(page_num: int): result.processing_time = time() - start_time return result - def _process_page(self, page, image_dir: Optional[Path]) -> PDFPage: + def _process_page(self, page, image_dir: Optional[Path], page_number: int) -> PDFPage: pdf_page = PDFPage( - page_number=self.current_page_number, + page_number=page_number, ) # Text and font extraction @@ -240,18 +237,18 @@ def visitor_text(text, cm, tm, font_dict, font_size): # Image extraction if self.extract_images: - pdf_page.images = self._extract_images(page, image_dir) + pdf_page.images = self._extract_images(page, image_dir, page_number) # Link extraction pdf_page.links = self._extract_links(page) # Add markdown content - pdf_page.markdown = clean_pdf_text(self.current_page_number, pdf_page.raw_text) - pdf_page.html = clean_pdf_text_to_html(self.current_page_number, pdf_page.raw_text) + pdf_page.markdown = clean_pdf_text(page_number, pdf_page.raw_text) + pdf_page.html = clean_pdf_text_to_html(page_number, pdf_page.raw_text) return pdf_page - def _extract_images(self, page, image_dir: Optional[Path]) -> List[Dict]: + def _extract_images(self, page, image_dir: Optional[Path], page_number: int) -> List[Dict]: # Import pypdf for type checking only when needed try: from pypdf.generic import IndirectObject @@ -276,7 +273,7 @@ def _extract_images(self, page, image_dir: Optional[Path]) -> List[Dict]: if xobj.get('/Subtype') == '/Image': try: img_count += 1 - img_filename = f"page_{self.current_page_number}_img_{img_count}" + img_filename = f"page_{page_number}_img_{img_count}" data = xobj.get_data() filters = xobj.get('/Filter', []) if not isinstance(filters, list): diff --git a/tests/unit/test_pdf_page_number_race.py b/tests/unit/test_pdf_page_number_race.py new file mode 100644 index 000000000..110fbd77e --- /dev/null +++ b/tests/unit/test_pdf_page_number_race.py @@ -0,0 +1,174 @@ +"""Regression tests for unclecode/crawl4ai#2319. + +`NaivePDFProcessorStrategy.process_batch()` used to keep the page number being +processed in the shared `self.current_page_number` attribute: each worker thread +wrote its own page number there, then read it back after text extraction. When +another worker overwrote it in between, the page-1-only formatting in +`clean_pdf_text()`/`clean_pdf_text_to_html()` (e.g. Title-Case author-line +bolding), the `PDFPage.page_number` field, and the extracted image file names +(`page_{n}_img_{m}`) all landed on the wrong page -- so the same PDF produced +different markdown on every run with the default `batch_size=4`. + +The fix passes the page number down as an explicit argument to `_process_page()` +and `_extract_images()` instead of storing it on `self`. These tests pin that: + +* `process_batch()` output is byte-identical across runs with `batch_size > 1` +* the page-1-only formatting applies to page 1 and only page 1 +* `PDFPage.page_number` is sequential even under the thread pool +* the serial `process()` and parallel `process_batch()` paths agree + +The test PDFs are hand-crafted minimal files so the tests need no extra +dependencies beyond pypdf (fpdf2 is not required). +""" + +import pytest + +pypdf = pytest.importorskip("pypdf") + +from crawl4ai.processors.pdf.processor import NaivePDFProcessorStrategy + +N_PAGES = 12 +BATCH_SIZE = 4 +TITLE_CASE_LINE = "Assessment Overview" + + +def _escape_pdf_text(text: str) -> str: + return text.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)") + + +def _page_lines(page_number: int): + return [ + f"Chapter Heading Number {page_number}", + TITLE_CASE_LINE, + f"this is the body of page {page_number}. " * 20, + ] + + +def _build_pdf(path, n_pages: int = N_PAGES) -> None: + """Write a minimal but valid multi-page PDF with plain-text content.""" + objects = [] + kids = " ".join(f"{3 + 3 * i} 0 R" for i in range(n_pages)) + objects.append("<< /Type /Catalog /Pages 2 0 R >>") + objects.append(f"<< /Type /Pages /Kids [{kids}] /Count {n_pages} >>") + for i in range(n_pages): + page_obj = 3 + 3 * i + content_obj = page_obj + 1 + font_obj = page_obj + 2 + objects.append( + f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] " + f"/Contents {content_obj} 0 R " + f"/Resources << /Font << /F1 {font_obj} 0 R >> >> >>" + ) + stream_lines = ["BT /F1 12 Tf 72 720 Td"] + for j, line in enumerate(_page_lines(i + 1)): + if j: + stream_lines.append("0 -16 Td") + stream_lines.append(f"({_escape_pdf_text(line)}) Tj") + stream_lines.append("ET") + stream = "\n".join(stream_lines).encode("latin-1") + objects.append((f"<< /Length {len(stream)} >>\nstream\n", stream, "\nendstream")) + objects.append("<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>") + + out = bytearray(b"%PDF-1.4\n") + offsets = [] + for idx, obj in enumerate(objects, start=1): + offsets.append(len(out)) + out += f"{idx} 0 obj\n".encode("latin-1") + if isinstance(obj, tuple): + out += obj[0].encode("latin-1") + obj[1] + obj[2].encode("latin-1") + else: + out += obj.encode("latin-1") + out += b"\nendobj\n" + xref_pos = len(out) + out += f"xref\n0 {len(objects) + 1}\n".encode("latin-1") + out += b"0000000000 65535 f \n" + for off in offsets: + out += f"{off:010d} 00000 n \n".encode("latin-1") + out += ( + f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\n" + f"startxref\n{xref_pos}\n%%EOF\n" + ).encode("latin-1") + path.write_bytes(bytes(out)) + + +def _strategy(**kwargs): + return NaivePDFProcessorStrategy(extract_images=False, **kwargs) + + +def _bolded_pages(result): + return tuple( + page.page_number + for page in result.pages + if f"**{TITLE_CASE_LINE}**" in page.markdown + ) + + +def test_process_batch_is_deterministic(tmp_path): + """Same PDF + batch_size > 1 must give byte-identical output every run.""" + pdf_path = tmp_path / "determinism.pdf" + _build_pdf(pdf_path) + + runs = [] + for _ in range(10): + result = _strategy(batch_size=BATCH_SIZE).process_batch(pdf_path) + runs.append(tuple(page.markdown for page in result.pages)) + + assert len(set(runs)) == 1, "process_batch output varied between runs (#2319)" + + +def test_page_one_only_formatting_in_batch(tmp_path): + """The page-1-only author-line bolding must apply to page 1, never others.""" + pdf_path = tmp_path / "page_one.pdf" + _build_pdf(pdf_path) + + for _ in range(5): + result = _strategy(batch_size=BATCH_SIZE).process_batch(pdf_path) + assert _bolded_pages(result) == (1,), ( + "page-1-only formatting leaked onto another page (#2319)" + ) + + +def test_page_numbers_sequential_in_batch(tmp_path): + """PDFPage.page_number must match the page's position, even under threads.""" + pdf_path = tmp_path / "numbering.pdf" + _build_pdf(pdf_path, n_pages=8) + + result = _strategy(batch_size=BATCH_SIZE).process_batch(pdf_path) + assert [page.page_number for page in result.pages] == list(range(1, 9)) + + +@pytest.mark.parametrize("page_number", [1, 2, 3]) +def test_process_page_uses_given_page_number(page_number): + """_process_page() must honor its page_number argument, not shared state.""" + + class _StubPage: + def extract_text(self, visitor_text=None, **kwargs): + text = "\n".join(_page_lines(page_number)) + if visitor_text is not None: + visitor_text(text, None, [0, 0, 0, 0, 0, 0], None, None) + return text + + def __contains__(self, key): # _extract_links checks '/Annots' in page + return False + + strategy = _strategy() + page = strategy._process_page(_StubPage(), None, page_number) + + assert page.page_number == page_number + bolded = f"**{TITLE_CASE_LINE}**" in page.markdown + assert bolded == (page_number == 1), ( + f"page-1-only formatting wrong for page_number={page_number}" + ) + + +def test_process_and_process_batch_agree(tmp_path): + """Serial and batched paths must produce identical per-page markdown/html.""" + pdf_path = tmp_path / "parity.pdf" + _build_pdf(pdf_path) + + serial = _strategy().process(pdf_path) + batched = _strategy(batch_size=BATCH_SIZE).process_batch(pdf_path) + + assert [p.markdown for p in serial.pages] == [p.markdown for p in batched.pages] + assert [p.html for p in serial.pages] == [p.html for p in batched.pages] + assert [p.page_number for p in serial.pages] == [p.page_number for p in batched.pages]