Stream PDF from archives - #117
Merged
Merged
Conversation
Add process_archive(), used automatically when --input points to a
.zip/.tar/.tar.gz (.tgz/.tar.bz2/.tbz2) archive. Eligible entries are read out
of the archive in chunks of batch_size, each chunk is extracted to a temporary
directory, sent to GROBID via the existing process_batch (so concurrency,
TEI/JSON/Markdown output, --force/skip and error handling are reused), and the
temporary files are removed before the next chunk is extracted. The archive is
never fully decompressed, so disk usage stays bounded regardless of its size.
- zip via zipfile, tar/tar.gz/tar.bz2 via tarfile ('r:*')
- entries are streamed member-by-member (zipfile.open / tarfile.extractfile)
- archive paths are sanitized to prevent path traversal (zip slip)
- when --output is omitted, results go to a directory named after the archive
- directory-input eligibility check refactored into _is_eligible_input and shared
Documented in the Readme and covered by unit tests (zip, tar.gz, chunking,
cleanup, default output, traversal guard, delegation).
…rsal --input now accepts shell-style glob patterns (with recursive **), e.g. 'paper.zip' (one file), 'paper*.zip' (many), '**/paper*.zip' (subdirectories) or '**/*.pdf'. Each match is dispatched by type: archives are streamed, directories are recursed, eligible files are processed directly, and the results of all matches are aggregated into a single summary. - resolve --input via glob (has_magic + recursive=True, ~ expansion); a plain path is returned unchanged for backward compatibility - directory traversal refactored from os.walk to pathlib.Path.rglob - factor the batching loop, stats summary and archive streaming into reusable helpers (_run_file_batches, _print_processing_summary, _process_archive_core) so directory, loose-file and archive inputs share one code path - loose files matched by a glob are batched together under their common base Documented in the Readme and covered by unit tests (multi-archive glob, recursive **/*.pdf, mixed matches, no-match warning, common-base helper).
lfoppiano
force-pushed
the
feature/stream-archive-input
branch
from
August 11, 2026 21:01
5400cf2 to
87b7e95
Compare
Add s3:// support to --input (and a new --input-list manifest). An s3 zip is range-streamed with smart_open (only the central directory and the requested entries are fetched - the object is never fully downloaded); loose remote PDFs are fetched a batch at a time to a temp dir. Mixed manifests (local + glob + s3) are supported and aggregated into one summary. - refactor process() -> process_paths(list of inputs); process() delegates - _resolve_input_paths handles s3 object / prefix / glob (list_objects_v2 + fnmatch) - _open_archive range-streams s3 zips (smart_open seekable stream); the stream is closed after use; s3 tar is rejected (not range-streamable) - new _process_remote_files streams loose s3 objects in bounded chunks - --input-list reads a file of paths (local/glob/s3, '#' comments) - s3 deps (smart_open[s3], boto3) are an optional 'pip install ...[s3]' extra, lazily imported with a clear install hint if missing - credentials use the standard AWS chain (env / ~/.aws / IAM) Documented in the Readme; covered by moto-backed tests (single object, prefix, glob, zip range-streaming, loose PDFs, mixed manifest, missing-extra error).
process_batch decides a document is already done with os.path.isfile() alone,
so any output truncated by a kill is indistinguishable from a complete one and
is skipped on every subsequent run. The corruption is permanent and silent: it
survives every resume, and `find -name '*.grobid.tei.xml' | wc -l` counts it as
a success. At corpus scale, on jobs that routinely hit a wall clock or a memory
limit, this is not a rare case.
_write_atomic writes to a temp file and renames, so the destination either does
not exist or is the whole document. Applied to all six write sites: TEI, the
error file, and the JSON/Markdown pairs on both the fresh and already-exists
paths.
Three details that are easy to get wrong:
- The temp file goes in the DESTINATION directory, not TMPDIR. os.replace is
only atomic within a filesystem, and on a cluster TMPDIR is usually a
different mount.
- mkstemp hardcodes 0600, where open() respects the umask. Renaming such a
file into place would leave every output private -- unreadable to the group
on shared scratch, which is where the results of a cluster run live. The
mode is taken from a umask read once at import, since reading it means
temporarily setting it and that is not safe from worker threads.
- fd ownership transfers to the file object on a successful fdopen, so the
error path closes the descriptor only when fdopen itself failed. Closing it
unconditionally could close an unrelated descriptor that reused the number,
and process_batch writes from a ThreadPoolExecutor.
The "." prefix and ".tmp" suffix keep an in-flight temp file from matching
*.grobid.tei.xml or *_[0-9]*.txt, so output counting is unaffected mid-write;
there is a test pinning that. The other tests cover the case that matters: a
writer that puts bytes on disk and then dies leaves no destination file and no
temp file, and a failed overwrite preserves the previous content.
Residual risk: a SIGKILL between mkstemp and rename leaks a temp file. That is
visible and harmless, unlike a truncated TEI.
The package ships a py.typed marker since #112, which promises inline types to type checkers of anything importing it; the methods added by this branch had none, so process_paths, process_archive and the S3 helpers came back untyped to every user of the client. The signatures now match the ones on master. _open_archive's handle stays Any on purpose: a ZipFile and a TarFile share no interface here, which is exactly why the function returns a "kind" tag for the callers to dispatch on. mypy --ignore-missing-imports is clean on the module, as it is on master.
lfoppiano
force-pushed
the
feature/stream-archive-input
branch
from
August 11, 2026 21:12
87b7e95 to
ac5b7db
Compare
lfoppiano
added a commit
that referenced
this pull request
Aug 15, 2026
The archive and s3 streaming (#117) shipped with a known detour: every entry was written to a temporary directory only so that process_pdf could open it again from a path, with the commit itself noting this would go away once PR #67 landed. It has landed, so this plugs the two together: archive entries and loose s3 objects are now read straight into memory and posted from there, named after the entry (or the s3 basename), and nothing but the results ever touches the disk. process_batch accepts the in-memory documents alongside paths - an entry goes by the name it carries, and since process_pdf returns that same name, the result lands on the same output file it would have as a path. The one input that still takes the temp-dir route is processCitationList, whose .txt files are read by process_txt from a path. The archive tests asserted on the temp dirs the posts came from, which no longer exist; they now assert on what actually crossed the wire - each entry posted once, under its archive name, with its own bytes - plus explicitly that mkdtemp is never called on the pdf path. Completes what #117 left pending on #67.
lfoppiano
added a commit
that referenced
this pull request
Aug 16, 2026
process_pdf could only read the document from disk, so callers holding a PDF in memory - fetched from an API, read out of a database or an object store - had to write it to a temporary file only for the client to open it again. It now takes the document itself as well: bytes, or any binary stream. Nothing says which of the two it is; the object does. A document also names itself, from the "name" attribute open() sets on files and that can be set on anything else, io.BytesIO included, so the identity of a document is not lost by going through memory - it travels with the request and comes back with the result. Bytes on their own have nothing to be named after and fall back to DEFAULT_IN_MEMORY_NAME. A stream is read once, up front, and re-served from memory afterwards: the 503 retry sends the same document again, and a consumed (or non-seekable) stream would silently post an empty body the second time around. That is also why the retry no longer recurses through the public entry point, which would have had to re-derive a name from a source that is by then exhausted. process_documents processes several of them concurrently, through the same ThreadPoolExecutor the file-based processing uses. Results come back in input order rather than in completion order: in-memory documents have no filenames to be matched back on afterwards, so the caller has nothing but the order to zip them onto. A single PDF passed by mistake raises instead of being iterated, which would otherwise send one request per byte. An in-memory run keeps up to n documents in flight against the server, so a client concurrency above the server's engine pool only piles up requests that queue there or come back as 503, while one below it leaves engines idle. Neither is visible from the client side until the throughput disappoints. Before process_documents and the in-memory archive/s3 streaming start, the client now asks /api/health how many engines the server has (pool.maxActive) and logs a warning when n exceeds them - with the number to use instead - and an info message when they outnumber n. The check is advisory, not a gate: a server without the endpoint (older GROBID), an unreadable answer or a connection failure never blocks the run. A server answering ready: false is also surfaced as a warning. Completes what #117 left pending on #67. --------- Co-authored-by: Jan Göpfert <94385965+jangoepfert@users.noreply.github.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.