From 881974b8a95596cb2913dd9138bd10d189de4927 Mon Sep 17 00:00:00 2001 From: Peter Argo Date: Fri, 10 Jul 2026 01:04:07 -0400 Subject: [PATCH] Add batch ingestion with duplicate skipping and parallel parsing New docex/batch module: BatchIngestor files every document in a folder into a basket in one call and returns a BatchReport with per-file outcomes. Files already in the basket are skipped using the same fingerprint rule as add(), so re-running over the same folder only picks up new files. Per-file errors are recorded in the report instead of aborting the batch. An optional processor runs on each added document; processors that separate parsing from persistence get their parsing parallelized in a thread pool while all database writes stay single-threaded. Includes tests and an example. Co-authored-by: Cursor --- docex/batch/__init__.py | 25 ++++ docex/batch/ingest.py | 202 ++++++++++++++++++++++++++++ examples/batch_ingestion_example.py | 62 +++++++++ tests/batch/__init__.py | 0 tests/batch/conftest.py | 26 ++++ tests/batch/test_batch_ingest.py | 139 +++++++++++++++++++ 6 files changed, 454 insertions(+) create mode 100644 docex/batch/__init__.py create mode 100644 docex/batch/ingest.py create mode 100644 examples/batch_ingestion_example.py create mode 100644 tests/batch/__init__.py create mode 100644 tests/batch/conftest.py create mode 100644 tests/batch/test_batch_ingest.py diff --git a/docex/batch/__init__.py b/docex/batch/__init__.py new file mode 100644 index 0000000..cacdeb2 --- /dev/null +++ b/docex/batch/__init__.py @@ -0,0 +1,25 @@ +""" +Batch ingestion for DocEX: file a folder of documents in one call. + +See :class:`docex.batch.ingest.BatchIngestor`. Duplicate files are skipped +(safe to re-run), per-file errors are collected instead of raised, and content +parsing is parallelized while database writes stay single-threaded. +""" + +from docex.batch.ingest import ( + ADDED, + FAILED, + SKIPPED_DUPLICATE, + BatchIngestor, + BatchReport, + FileOutcome, +) + +__all__ = [ + 'BatchIngestor', + 'BatchReport', + 'FileOutcome', + 'ADDED', + 'SKIPPED_DUPLICATE', + 'FAILED', +] diff --git a/docex/batch/ingest.py b/docex/batch/ingest.py new file mode 100644 index 0000000..22ad8fa --- /dev/null +++ b/docex/batch/ingest.py @@ -0,0 +1,202 @@ +""" +Batch ingestion: file a whole folder of documents in one call and get a report. + +Design notes: + +* Files already in the basket (same content fingerprint and source path, the + same rule DocEX's add() uses for duplicates) are skipped, so re-running a + batch over the same folder is safe and only picks up new files. +* Database writes are deliberately single-threaded: DocEX's connection + handling is not thread-safe and SQLite serializes writers anyway. Only the + expensive, database-free work -- reading and parsing document content -- is + parallelized across a thread pool. +* Per-file errors are recorded in the report instead of raised, so one bad + file never aborts the batch. +""" + +import hashlib +import logging +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, List, Optional, Tuple + +logger = logging.getLogger(__name__) + +# File outcome statuses +ADDED = 'added' +SKIPPED_DUPLICATE = 'skipped_duplicate' +FAILED = 'failed' + +# The pure (no database writes) processor methods used for the parallel stage. +_SPLIT_PIPELINE_METHODS = ('read_text', 'extract_from_text', 'save_results') + + +@dataclass +class FileOutcome: + """What happened to one file in the batch.""" + + path: str + status: str + reason: Optional[str] = None + document_id: Optional[str] = None + needs_review: bool = False + + +@dataclass +class BatchReport: + """Result of a batch run: per-file outcomes plus convenience views.""" + + outcomes: List[FileOutcome] = field(default_factory=list) + + @property + def added(self) -> List[FileOutcome]: + return [o for o in self.outcomes if o.status == ADDED] + + @property + def skipped(self) -> List[FileOutcome]: + return [o for o in self.outcomes if o.status == SKIPPED_DUPLICATE] + + @property + def failed(self) -> List[FileOutcome]: + return [o for o in self.outcomes if o.status == FAILED] + + @property + def processing_failures(self) -> List[FileOutcome]: + return [o for o in self.outcomes if o.status == ADDED and o.reason] + + @property + def needs_review(self) -> List[FileOutcome]: + return [o for o in self.outcomes if o.needs_review] + + def summary(self) -> str: + parts = [ + f"{len(self.added)} added", + f"{len(self.skipped)} skipped (duplicates)", + f"{len(self.failed)} failed", + ] + if self.processing_failures: + parts.append(f"{len(self.processing_failures)} processing failures") + if self.needs_review: + parts.append(f"{len(self.needs_review)} need review") + return ', '.join(parts) + + +class BatchIngestor: + """Files every document in a folder into a basket, optionally running a processor.""" + + def __init__(self, basket: Any, processor: Optional[Any] = None, max_workers: int = 4): + """ + Args: + basket: DocBasket to file documents into. + processor: Optional DocEX processor to run on each added document. + A processor exposing read_text/extract_from_text/save_results + (such as the field extraction processor) gets its content + parsing parallelized; any other processor is run serially via + its standard process() method. + max_workers: Thread pool size for the parallel parsing stage. + """ + self.basket = basket + self.processor = processor + self.max_workers = max_workers + + def ingest_folder(self, folder: str, pattern: str = '*') -> BatchReport: + """Ingest every file in a folder (non-recursive) and return a report. + + Args: + folder: Path to the folder to ingest. + pattern: Optional glob pattern to filter files (e.g. '*.pdf'). + """ + folder_path = Path(folder) + if not folder_path.is_dir(): + raise ValueError(f"Not a folder: {folder}") + + report = BatchReport() + added: List[Tuple[FileOutcome, Any]] = [] + + # Phase 1 (serial): fingerprint, skip duplicates, file the rest. + for path in sorted(p for p in folder_path.glob(pattern) if not p.is_dir()): + outcome, document = self._ingest_file(path) + report.outcomes.append(outcome) + if document is not None: + added.append((outcome, document)) + + # Phase 2: run the processor on newly added documents. + if self.processor and added: + self._process_documents(added) + + logger.info(f"Batch ingest of {folder}: {report.summary()}") + return report + + def _ingest_file(self, path: Path) -> Tuple[FileOutcome, Optional[Any]]: + try: + checksum = hashlib.sha256(path.read_bytes()).hexdigest() + except OSError as e: + return FileOutcome(str(path), FAILED, reason=f"unreadable: {e}"), None + + if self._already_in_basket(checksum, str(path)): + return FileOutcome(str(path), SKIPPED_DUPLICATE), None + + try: + document = self.basket.add(str(path)) + except Exception as e: + return FileOutcome(str(path), FAILED, reason=f"add failed: {e}"), None + + return FileOutcome(str(path), ADDED, document_id=document.id), document + + def _already_in_basket(self, checksum: str, source: str) -> bool: + """Same duplicate rule as DocBasket.add(): matching fingerprint and source path.""" + from docex.db.models import Document as DocumentModel + + with self.basket.db.session() as session: + existing = ( + session.query(DocumentModel) + .filter_by(basket_id=self.basket.id, checksum=checksum, source=source) + .first() + ) + return existing is not None + + def _process_documents(self, added: List[Tuple[FileOutcome, Any]]) -> None: + if all(hasattr(self.processor, m) for m in _SPLIT_PIPELINE_METHODS): + self._process_split(added) + else: + self._process_serial(added) + + def _process_split(self, added: List[Tuple[FileOutcome, Any]]) -> None: + """Parallel parsing, serial persistence. + + Workers only read content and extract fields (no database writes); + results are saved by this (main) thread one document at a time. + """ + def parse(document): + return self.processor.extract_from_text(self.processor.read_text(document)) + + with ThreadPoolExecutor(max_workers=self.max_workers) as executor: + futures = { + executor.submit(parse, document): (outcome, document) + for outcome, document in added + } + for future in as_completed(futures): + outcome, document = futures[future] + try: + results = future.result() + except Exception as e: + outcome.reason = f"processing failed: {e}" + outcome.needs_review = True + continue + metadata = self.processor.save_results(document, results) + outcome.needs_review = metadata.get('needs_review') == 'true' + + def _process_serial(self, added: List[Tuple[FileOutcome, Any]]) -> None: + for outcome, document in added: + try: + result = self.processor.process(document) + except Exception as e: + outcome.reason = f"processing failed: {e}" + outcome.needs_review = True + continue + if not getattr(result, 'success', False): + outcome.reason = f"processing failed: {getattr(result, 'error', 'unknown error')}" + outcome.needs_review = True + else: + outcome.needs_review = (result.metadata or {}).get('needs_review') == 'true' diff --git a/examples/batch_ingestion_example.py b/examples/batch_ingestion_example.py new file mode 100644 index 0000000..8211434 --- /dev/null +++ b/examples/batch_ingestion_example.py @@ -0,0 +1,62 @@ +""" +Example: Batch ingestion of a folder of documents. + +Creates a small inbox folder, files everything in it into a basket in one +call, and prints the report. Run it twice to see duplicate skipping: the +second run adds nothing, so a nightly job can point at the same folder and +only pick up new files. + +To also extract fields from each document as it is filed, pass any DocEX +processor: + + ingestor = BatchIngestor(basket, processor=my_processor) + +Processors that separate parsing from persistence (read_text / +extract_from_text / save_results) get their parsing parallelized across a +thread pool; database writes always stay on a single thread. + +Note: DocEX must be initialized first using the CLI command 'docex init'. +""" + +import logging +from pathlib import Path + +from docex import DocEX +from docex.batch import BatchIngestor +from docex.context import UserContext + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +SAMPLE_INVOICES = { + 'invoice_april.txt': "Invoice No.: 2024-0104\nTotal Due: $12,500.00\n", + 'invoice_may.txt': "Invoice No.: 2024-0105\nTotal Due: $12,500.00\n", + 'invoice_june.txt': "Invoice No.: 2024-0106\nTotal Due: $13,100.00\n", +} + + +def main(): + inbox = Path('examples/sample_data/batch_inbox') + inbox.mkdir(parents=True, exist_ok=True) + for name, content in SAMPLE_INVOICES.items(): + (inbox / name).write_text(content) + + user_context = UserContext(user_id='batch_example', user_email='example@example.com') + docEX = DocEX(user_context=user_context) + basket = docEX.basket('batch_example') + + ingestor = BatchIngestor(basket) + + report = ingestor.ingest_folder(str(inbox), pattern='*.txt') + logger.info(f"First run: {report.summary()}") + for outcome in report.outcomes: + logger.info(f" {outcome.status:18} {Path(outcome.path).name} {outcome.reason or ''}") + + # Second run over the same folder: everything is recognized and skipped. + report = ingestor.ingest_folder(str(inbox), pattern='*.txt') + logger.info(f"Second run: {report.summary()}") + logger.info(f"Documents in basket: {basket.count_documents()}") + + +if __name__ == '__main__': + main() diff --git a/tests/batch/__init__.py b/tests/batch/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/batch/conftest.py b/tests/batch/conftest.py new file mode 100644 index 0000000..f82fdb6 --- /dev/null +++ b/tests/batch/conftest.py @@ -0,0 +1,26 @@ +"""Sandboxed DocEX environment for batch ingestion tests. + +DOCEX_HOME is pointed at a temporary directory before docex is imported, so +these tests never touch a real ~/.docex configuration or database. +""" + +import os +import tempfile +from pathlib import Path + +import pytest + +_SANDBOX = Path(tempfile.mkdtemp(prefix='docex_batch_tests_')) +os.environ.setdefault('DOCEX_HOME', str(_SANDBOX)) + + +@pytest.fixture(scope='session') +def docex_instance(): + home = Path(os.environ['DOCEX_HOME']) + from docex import DocEX + + DocEX.setup( + database={'type': 'sqlite', 'sqlite': {'path': str(home / 'docex.db')}}, + storage={'type': 'filesystem', 'filesystem': {'path': str(home / 'storage')}}, + ) + return DocEX() diff --git a/tests/batch/test_batch_ingest.py b/tests/batch/test_batch_ingest.py new file mode 100644 index 0000000..30eba32 --- /dev/null +++ b/tests/batch/test_batch_ingest.py @@ -0,0 +1,139 @@ +"""Batch ingestion: counts, duplicate skipping, error capture, processor stages.""" + +import threading + +import pytest + +from docex.batch import BatchIngestor +from docex.processors.base import BaseProcessor, ProcessingResult +from docex.services.metadata_service import MetadataService + + +@pytest.fixture() +def basket(docex_instance, request): + return docex_instance.basket(f'batch_test_{request.node.name}') + + +@pytest.fixture() +def invoice_folder(tmp_path): + folder = tmp_path / 'inbox' + folder.mkdir() + for i in range(3): + (folder / f'invoice_{i}.txt').write_text(f"Invoice No.: 2024-000{i}\nTotal Due: $1,{i}00.00\n") + return folder + + +class SplitStubProcessor(BaseProcessor): + """Stub with the split pipeline API (parallel parse stage, serial save stage).""" + + def __init__(self, db=None, fail_on=None): + super().__init__({}, db) + self.fail_on = fail_on or set() + self.parse_threads = set() + self.save_threads = set() + + def can_process(self, document): + return True + + def read_text(self, document): + if document.name in self.fail_on: + raise ValueError('cannot parse this document') + return self.get_document_text(document) + + def extract_from_text(self, text): + self.parse_threads.add(threading.current_thread().name) + return {'parsed_chars': str(len(text))} + + def save_results(self, document, results): + self.save_threads.add(threading.current_thread().name) + metadata = dict(results) + metadata['needs_review'] = 'false' + MetadataService(self.db).update_metadata(document.id, metadata) + return metadata + + def process(self, document): + return ProcessingResult(success=True, metadata=self.save_results(document, self.extract_from_text(self.read_text(document)))) + + +class PlainStubProcessor(BaseProcessor): + """Stub with only the standard process() method (generic serial path).""" + + def can_process(self, document): + return True + + def process(self, document): + metadata = {'needs_review': 'true'} + MetadataService(self.db).update_metadata(document.id, metadata) + return ProcessingResult(success=True, metadata=metadata) + + +def test_folder_ingested_and_rerun_skips_everything(basket, invoice_folder): + ingestor = BatchIngestor(basket) + + first = ingestor.ingest_folder(str(invoice_folder)) + assert len(first.added) == 3 + assert len(first.skipped) == 0 + assert len(first.failed) == 0 + + # Re-running the same folder is idempotent: nothing filed twice. + second = ingestor.ingest_folder(str(invoice_folder)) + assert len(second.added) == 0 + assert len(second.skipped) == 3 + assert basket.count_documents() == 3 + + +def test_new_file_picked_up_on_rerun(basket, invoice_folder): + ingestor = BatchIngestor(basket) + ingestor.ingest_folder(str(invoice_folder)) + + (invoice_folder / 'invoice_late.txt').write_text("Invoice No.: 2024-9999\n") + report = ingestor.ingest_folder(str(invoice_folder)) + assert len(report.added) == 1 + assert report.added[0].path.endswith('invoice_late.txt') + + +def test_unreadable_file_recorded_not_raised(basket, invoice_folder, tmp_path): + (invoice_folder / 'broken.txt').symlink_to(tmp_path / 'does_not_exist.txt') + + report = BatchIngestor(basket).ingest_folder(str(invoice_folder)) + assert len(report.added) == 3 + assert len(report.failed) == 1 + assert 'unreadable' in report.failed[0].reason + + +def test_split_processor_parses_in_workers_and_saves_in_main_thread(basket, invoice_folder): + processor = SplitStubProcessor() + report = BatchIngestor(basket, processor=processor, max_workers=3).ingest_folder(str(invoice_folder)) + + assert len(report.added) == 3 + # Parsing ran in worker threads; all database writes stayed on the main thread. + assert all('ThreadPoolExecutor' in name for name in processor.parse_threads) + assert processor.save_threads == {threading.main_thread().name} + # Extraction results were persisted as metadata. + doc = basket.get_document(report.added[0].document_id) + assert 'parsed_chars' in doc.get_metadata() + + +def test_processing_failure_flags_file_and_batch_continues(basket, invoice_folder): + processor = SplitStubProcessor(fail_on={'invoice_1.txt'}) + report = BatchIngestor(basket, processor=processor).ingest_folder(str(invoice_folder)) + + assert len(report.added) == 3 + assert len(report.processing_failures) == 1 + assert 'processing failed' in report.processing_failures[0].reason + assert report.processing_failures[0].needs_review + + +def test_plain_processor_runs_serially_and_review_flags_reported(basket, invoice_folder): + report = BatchIngestor(basket, processor=PlainStubProcessor({})).ingest_folder(str(invoice_folder)) + + assert len(report.added) == 3 + assert len(report.needs_review) == 3 + assert 'need review' in report.summary() + + +def test_glob_pattern_filters_files(basket, invoice_folder): + (invoice_folder / 'notes.md').write_text('not an invoice') + + report = BatchIngestor(basket).ingest_folder(str(invoice_folder), pattern='*.txt') + assert all(outcome.path.endswith('.txt') for outcome in report.outcomes)