From 2dbe67b92a9505f19830a797fba0880bbe0247f4 Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Mon, 10 Aug 2026 17:33:00 +0200 Subject: [PATCH] Add --skip_errors to skip documents that failed in a previous run By default a re-run only skips documents whose TEI output already exists, so documents that failed are sent to GROBID again even though, unless something changed, they will fail again. --skip_errors (skip_errors=... in the library) also skips the documents for which a previous run left an error file _.txt next to the expected TEI output. --force still reprocesses everything. Since the error file now drives a decision and not only reporting, it is kept in sync: it is removed once the document is processed successfully, and a failure with a different status replaces the previous marker instead of accumulating next to it. Closes #119 --- Readme.md | 39 +++++++-- grobid_client/grobid_client.py | 139 ++++++++++++++++++++++++++----- tests/test_grobid_client.py | 145 +++++++++++++++++++++++++++++++++ 3 files changed, 292 insertions(+), 31 deletions(-) diff --git a/Readme.md b/Readme.md index d6a74ce..fdf4a2e 100644 --- a/Readme.md +++ b/Readme.md @@ -126,15 +126,16 @@ grobid_client [OPTIONS] SERVICE #### Common Options -| Option | Description | Default | -|-------------|--------------------------|-------------------------| -| `--input` | Input directory path | Required | -| `--output` | Output directory path | Same as input | -| `--server` | GROBID server URL | `http://localhost:8070` | -| `--n` | Concurrency level | 10 | -| `--config` | Config file path | Optional | -| `--force` | Overwrite existing files | False | -| `--verbose` | Enable verbose logging | False | +| Option | Description | Default | +|-----------------|---------------------------------------------------|-------------------------| +| `--input` | Input directory path | Required | +| `--output` | Output directory path | Same as input | +| `--server` | GROBID server URL | `http://localhost:8070` | +| `--n` | Concurrency level | 10 | +| `--config` | Config file path | Optional | +| `--force` | Overwrite existing files | False | +| `--skip_errors` | Also skip documents that failed in a previous run | False | +| `--verbose` | Enable verbose logging | False | #### Processing Options @@ -173,6 +174,9 @@ grobid_client --server https://grobid.example.com --input ~/citations.txt proces # Force reprocessing with sentence segmentation and JSON output grobid_client --input ~/docs --force --segment_sentences --json processFulltextDocument +# Resume an interrupted run without retrying the documents that already failed +grobid_client --input ~/docs --output ~/results --skip_errors processFulltextDocument + # Process PDFs directly from a zip or tar.gz archive (streamed, not fully decompressed) grobid_client --input ~/papers.zip --output ~/results processFulltextDocument grobid_client --input ~/papers.tar.gz --output ~/results processFulltextDocument @@ -203,6 +207,14 @@ grobid_client --input "~/data/**/*.pdf" --output ~/results processFulltextDocu > A **manifest of paths** (local, glob or `s3://`, one per line, `#` comments allowed) can be processed together via > `--input-list paths.txt` (combinable with `--input`). +> [!NOTE] +> **Skipping already handled documents.** By default a re-run skips a document only when its TEI output already exists, +> so documents that failed are sent to GROBID again. Since a failed document generally fails again unless something +> changed, `--skip_errors` also skips the documents for which a previous run left an error file +> (`_.txt`, e.g. `paper_500.txt`) next to the expected TEI output. Drop the flag (or use `--force`) to +> retry them. Error files are kept in sync automatically: the marker is deleted once the document is processed +> successfully, and replaced when the same document fails again with a different status code. + ### Python Library #### Basic Usage @@ -259,6 +271,15 @@ client.process( markdown_output=True ) +# Re-run without retrying the documents that failed before +client.process( + service="processFulltextDocument", + input_path="/path/to/pdfs", + output_path="/path/to/output", + force=False, + skip_errors=True +) + ```python # Process citation lists client.process( diff --git a/grobid_client/grobid_client.py b/grobid_client/grobid_client.py index 69b86fc..d32677a 100644 --- a/grobid_client/grobid_client.py +++ b/grobid_client/grobid_client.py @@ -75,6 +75,13 @@ class GrobidClient(ApiClient): # come before their single-dot prefixes when stripping (see _archive_stem). ARCHIVE_EXTENSIONS = (".tar.gz", ".tar.bz2", ".tgz", ".tbz2", ".zip", ".tar") + # Suffix of the TEI result files, and the naming of the error files written + # next to them when a document fails (e.g. paper_500.txt). The latter is what + # skip_errors looks for to know that a document already failed. + # See https://github.com/grobidOrg/grobid-client-python/issues/119 + TEI_SUFFIX = ".grobid.tei.xml" + ERROR_FILE_RE = re.compile(r"_\d{3}\.txt$") + # Default configuration values DEFAULT_CONFIG: dict = { 'grobid_server': 'http://localhost:8070', @@ -428,6 +435,45 @@ def _unlink_quietly(path: str) -> None: except OSError: pass + def _error_file_name(self, tei_filename: str, status: int) -> str: + """Name of the error file recording a failed processing of a document.""" + return self._output_stem(tei_filename) + f"_{status}.txt" + + def _output_stem(self, tei_filename: str) -> str: + """Strip the TEI suffix from an output file name.""" + if tei_filename.endswith(self.TEI_SUFFIX): + return tei_filename[:-len(self.TEI_SUFFIX)] + return tei_filename + + def _find_error_files(self, tei_filename: str) -> list: + """Return the error files left by previous failed runs for this output. + + Errors are recorded as ``_.txt`` next to the TEI file (the + status code varies from run to run), so we glob on the stem and keep only + the names that actually look like an error marker. + """ + stem = os.path.expanduser(self._output_stem(tei_filename)) + return [ + candidate for candidate in sorted(glob.glob(glob.escape(stem) + "_*.txt")) + if self.ERROR_FILE_RE.search(os.path.basename(candidate)) + ] + + def _remove_error_files(self, tei_filename: str, keep: Optional[str] = None) -> None: + """Delete stale error markers for an output, optionally keeping one. + + Called after a document has been (re)processed so that a document which + now succeeds - or fails with a different status - does not keep dragging + along the marker of an older failure. + """ + for error_file in self._find_error_files(tei_filename): + if keep is not None and os.path.abspath(error_file) == os.path.abspath(os.path.expanduser(keep)): + continue + try: + os.remove(error_file) + self.logger.debug(f"Removed stale error file: {error_file}") + except OSError as e: + self.logger.warning(f"Could not remove stale error file {error_file}: {str(e)}") + def ping(self) -> Tuple[bool, int]: """ Check the Grobid service. Returns True if the service is up. @@ -452,7 +498,8 @@ def process( verbose: bool = False, flavor: Optional[str] = None, json_output: bool = False, - markdown_output: bool = False + markdown_output: bool = False, + skip_errors: bool = False ) -> None: if input_path is None: self.logger.warning("No input path provided") @@ -465,6 +512,7 @@ def process( tei_coordinates=tei_coordinates, segment_sentences=segment_sentences, force=force, verbose=verbose, flavor=flavor, json_output=json_output, markdown_output=markdown_output, + skip_errors=skip_errors, ) def process_paths( @@ -484,7 +532,8 @@ def process_paths( verbose: bool = False, flavor: Optional[str] = None, json_output: bool = False, - markdown_output: bool = False + markdown_output: bool = False, + skip_errors: bool = False ) -> None: """Process a list of inputs. @@ -543,7 +592,8 @@ def process_paths( service, fs_files, self._common_base(fs_files), output, n, generate_ids, consolidate_header, consolidate_citations, include_raw_citations, include_raw_affiliations, tei_coordinates, - segment_sentences, force, verbose, flavor, json_output, markdown_output + segment_sentences, force, verbose, flavor, json_output, markdown_output, + skip_errors=skip_errors ) processed_files_count += bp errors_files_count += be @@ -556,7 +606,8 @@ def process_paths( service, remote_files, output, n, generate_ids, consolidate_header, consolidate_citations, include_raw_citations, include_raw_affiliations, tei_coordinates, - segment_sentences, force, verbose, flavor, json_output, markdown_output + segment_sentences, force, verbose, flavor, json_output, markdown_output, + skip_errors=skip_errors ) processed_files_count += rp errors_files_count += re_count @@ -569,7 +620,8 @@ def process_paths( service, archive_path, output, n, generate_ids, consolidate_header, consolidate_citations, include_raw_citations, include_raw_affiliations, tei_coordinates, - segment_sentences, force, verbose, flavor, json_output, markdown_output + segment_sentences, force, verbose, flavor, json_output, markdown_output, + skip_errors=skip_errors ) processed_files_count += ap errors_files_count += ae @@ -582,7 +634,8 @@ def process_paths( runtime = time.time() - start_time self._print_processing_summary( - processed_files_count, errors_files_count, skipped_files_count, total_files, runtime + processed_files_count, errors_files_count, skipped_files_count, total_files, runtime, + skip_errors=skip_errors ) def _resolve_input_paths(self, input_path: str) -> list: @@ -708,7 +761,8 @@ def _print_processing_summary( errors: int, skipped: int, total: int, - runtime: float + runtime: float, + skip_errors: bool = False ) -> None: """Print the final processing statistics (shared by all input modes).""" docs_per_second = processed / runtime if runtime > 0 else 0 @@ -717,7 +771,8 @@ def _print_processing_summary( print(f"Processing completed: {processed} out of {total} files processed") print(f"Errors: {errors} out of {total} files processed") if skipped > 0: - print(f"Skipped: {skipped} out of {total} files (already existed, use --force to reprocess)") + reason = "already existed or previously failed" if skip_errors else "already existed" + print(f"Skipped: {skipped} out of {total} files ({reason}, use --force to reprocess)") print(f"⏱️ Total runtime: {runtime:.2f} seconds") print(f"🚀 Speed: {docs_per_second:.2f} documents/second") @@ -741,7 +796,8 @@ def _run_file_batches( verbose: bool, flavor: Optional[str], json_output: bool, - markdown_output: bool + markdown_output: bool, + skip_errors: bool = False ) -> Tuple[int, int, int]: """Run process_batch over a list of files in chunks of batch_size. @@ -768,7 +824,8 @@ def _run_file_batches( service, batch, input_path, output, n, generate_ids, consolidate_header, consolidate_citations, include_raw_citations, include_raw_affiliations, tei_coordinates, segment_sentences, - force, verbose, flavor, json_output, markdown_output + force, verbose, flavor, json_output, markdown_output, + skip_errors=skip_errors ) processed_files_count += batch_processed errors_files_count += batch_errors @@ -780,7 +837,8 @@ def _run_file_batches( service, batch, input_path, output, n, generate_ids, consolidate_header, consolidate_citations, include_raw_citations, include_raw_affiliations, tei_coordinates, segment_sentences, - force, verbose, flavor, json_output, markdown_output + force, verbose, flavor, json_output, markdown_output, + skip_errors=skip_errors ) processed_files_count += batch_processed errors_files_count += batch_errors @@ -914,7 +972,8 @@ def process_archive( verbose: bool = False, flavor: Optional[str] = None, json_output: bool = False, - markdown_output: bool = False + markdown_output: bool = False, + skip_errors: bool = False ) -> None: """Process the eligible files contained in a zip/tar archive. @@ -933,14 +992,16 @@ def process_archive( service, archive_path, output, n, generate_ids, consolidate_header, consolidate_citations, include_raw_citations, include_raw_affiliations, tei_coordinates, segment_sentences, force, verbose, flavor, - json_output, markdown_output + json_output, markdown_output, skip_errors=skip_errors ) if total_files == 0: return runtime = time.time() - start_time - self._print_processing_summary(processed, errors, skipped, total_files, runtime) + self._print_processing_summary( + processed, errors, skipped, total_files, runtime, skip_errors=skip_errors + ) def _process_archive_core( self, @@ -959,7 +1020,8 @@ def _process_archive_core( verbose: bool, flavor: Optional[str], json_output: bool, - markdown_output: bool + markdown_output: bool, + skip_errors: bool = False ) -> Tuple[int, int, int, int]: """Stream and process an archive; return (total, processed, errors, skipped). @@ -1032,7 +1094,8 @@ def _process_archive_core( verbose, flavor, json_output, - markdown_output + markdown_output, + skip_errors=skip_errors ) processed_files_count += batch_processed errors_files_count += batch_errors @@ -1067,7 +1130,8 @@ def _process_remote_files( verbose: bool, flavor: Optional[str], json_output: bool, - markdown_output: bool + markdown_output: bool, + skip_errors: bool = False ) -> Tuple[int, int, int, int]: """Stream loose remote (s3) files to a temp dir in chunks and process them. @@ -1124,7 +1188,8 @@ def _process_remote_files( verbose, flavor, json_output, - markdown_output + markdown_output, + skip_errors=skip_errors ) processed_count += batch_processed error_count += batch_errors @@ -1152,7 +1217,8 @@ def process_batch( verbose: bool = False, flavor: Optional[str] = None, json_output: bool = False, - markdown_output: bool = False + markdown_output: bool = False, + skip_errors: bool = False ) -> Tuple[int, int, int]: batch_start_time = time.time() if verbose: @@ -1217,6 +1283,18 @@ def process_batch( continue + # Documents that already failed are usually going to fail again + # as long as nothing changed, so --skip-errors lets a re-run go + # straight past them. See issue #119. + if not force and skip_errors: + previous_errors = self._find_error_files(filename) + if previous_errors: + self.logger.info( + f"{input_file} previously failed ({os.path.basename(previous_errors[0])}), " + f"skipping... (use --force to retry it)") + skipped_count += 1 + continue + selected_process: Any = self.process_pdf if service == 'processCitationList': selected_process = self.process_txt @@ -1249,10 +1327,14 @@ def process_batch( self.logger.error(f"Processing of {input_file} failed with error {status}: {text}") error_count += 1 # writing error file with suffixed error code + error_filename = self._error_file_name(filename, status) try: - error_filename = filename.replace(".grobid.tei.xml", f"_{status}.txt") self._write_atomic(error_filename, text if text is not None else "") self.logger.info(f"Error details written to {error_filename}") + # A previous run may have failed with a different status code; + # keep only the marker of this run so the error state of the + # document stays unambiguous (and skippable). + self._remove_error_files(filename, keep=error_filename) except OSError as e: self.logger.error(f"Failed to write error file {filename}: {str(e)}") else: @@ -1261,7 +1343,12 @@ def process_batch( try: self._write_atomic(filename, text) self.logger.debug(f"Successfully wrote TEI file: {filename}") - + + # The document no longer is in error: drop the marker left by + # a previous failed run so it is not skipped by --skip-errors. + self._remove_error_files(filename) + + # Convert to JSON if requested if json_output: try: @@ -1554,6 +1641,12 @@ def main() -> None: action="store_true", help="force re-processing pdf input files when tei output files already exist", ) + parser.add_argument( + "--skip_errors", "--skip-errors", + dest="skip_errors", + action="store_true", + help="skip input files that already failed in a previous run (an error file _.txt exists next to the expected tei output); ignored when --force is used", + ) parser.add_argument( "--tei_coordinates", "--teiCoordinates", dest="tei_coordinates", @@ -1645,6 +1738,7 @@ def main() -> None: include_raw_citations = args.include_raw_citations include_raw_affiliations = args.include_raw_affiliations force = args.force + skip_errors = args.skip_errors tei_coordinates = args.tei_coordinates segment_sentences = args.segment_sentences verbose = args.verbose @@ -1691,7 +1785,8 @@ def main() -> None: verbose=verbose, flavor=flavor, json_output=json_output, - markdown_output=markdown_output + markdown_output=markdown_output, + skip_errors=skip_errors ) except Exception as e: logger.error(f"Processing failed: {str(e)}") diff --git a/tests/test_grobid_client.py b/tests/test_grobid_client.py index a46ecdd..54ddc15 100644 --- a/tests/test_grobid_client.py +++ b/tests/test_grobid_client.py @@ -1019,3 +1019,148 @@ def test_write_uses_normal_permissions(self): dest = os.path.join(d, 'a.grobid.tei.xml') client._write_atomic(dest, 'x') assert (os.stat(dest).st_mode & 0o777) == (os.stat(reference).st_mode & 0o777) + + +class TestSkipErrors: + """Tests for skipping documents that already failed (issue #119).""" + + def _client(self): + with patch('grobid_client.grobid_client.GrobidClient._test_server_connection'): + with patch('grobid_client.grobid_client.GrobidClient._configure_logging'): + client = GrobidClient(check_server=False) + client.logger = Mock() + client.config['batch_size'] = 50 + return client + + @staticmethod + def _make_pdfs(directory, names): + for name in names: + with open(os.path.join(directory, name), 'wb') as f: + f.write(b'%PDF') + + def _run(self, client, input_dir, output, failing=(), **kwargs): + """Process input_dir, making the named PDFs come back with a 500.""" + def fake_post(url, files=None, data=None, headers=None, timeout=None): + resp = Mock() + if os.path.basename(files['input'][0]) in failing: + resp.text = 'boom' + return (resp, 500) + resp.text = 'ok' + return (resp, 200) + + with patch.object(GrobidClient, 'post', side_effect=fake_post) as mock_post: + client.process('processFulltextDocument', input_dir, output=output, **kwargs) + return [os.path.basename(c.kwargs['files']['input'][0]) for c in mock_post.call_args_list] + + def test_find_error_files_matches_only_error_markers(self): + client = self._client() + with tempfile.TemporaryDirectory() as d: + tei = os.path.join(d, 'a.grobid.tei.xml') + for name in ('a_500.txt', 'a_408.txt', 'a_notes.txt', 'a.txt', 'b_500.txt'): + open(os.path.join(d, name), 'w').close() + found = [os.path.basename(f) for f in client._find_error_files(tei)] + assert found == ['a_408.txt', 'a_500.txt'] + + def test_error_file_name(self): + client = self._client() + assert client._error_file_name('/out/a.grobid.tei.xml', 500) == '/out/a_500.txt' + + def test_skip_errors_skips_previously_failed_document(self): + client = self._client() + with tempfile.TemporaryDirectory() as d: + inp = os.path.join(d, 'in') + out = os.path.join(d, 'out') + os.makedirs(inp) + self._make_pdfs(inp, ['ok.pdf', 'bad.pdf']) + + # first run: bad.pdf fails and leaves bad_500.txt behind + self._run(client, inp, out, failing={'bad.pdf'}, force=True) + assert os.path.isfile(os.path.join(out, 'ok.grobid.tei.xml')) + assert os.path.isfile(os.path.join(out, 'bad_500.txt')) + + # second run without --force: ok.pdf is skipped (tei exists) and + # bad.pdf is skipped too because of its error marker + sent = self._run(client, inp, out, failing={'bad.pdf'}, force=False, skip_errors=True) + assert sent == [] + + def test_without_skip_errors_failed_document_is_retried(self): + client = self._client() + with tempfile.TemporaryDirectory() as d: + inp = os.path.join(d, 'in') + out = os.path.join(d, 'out') + os.makedirs(inp) + self._make_pdfs(inp, ['ok.pdf', 'bad.pdf']) + + self._run(client, inp, out, failing={'bad.pdf'}, force=True) + # default behaviour is unchanged: only the existing tei is skipped + sent = self._run(client, inp, out, failing={'bad.pdf'}, force=False) + assert sent == ['bad.pdf'] + + def test_force_overrides_skip_errors(self): + client = self._client() + with tempfile.TemporaryDirectory() as d: + inp = os.path.join(d, 'in') + out = os.path.join(d, 'out') + os.makedirs(inp) + self._make_pdfs(inp, ['bad.pdf']) + + self._run(client, inp, out, failing={'bad.pdf'}, force=True) + sent = self._run(client, inp, out, failing={'bad.pdf'}, force=True, skip_errors=True) + assert sent == ['bad.pdf'] + + def test_successful_retry_removes_stale_error_file(self): + client = self._client() + with tempfile.TemporaryDirectory() as d: + inp = os.path.join(d, 'in') + out = os.path.join(d, 'out') + os.makedirs(inp) + self._make_pdfs(inp, ['bad.pdf']) + + self._run(client, inp, out, failing={'bad.pdf'}, force=True) + assert os.path.isfile(os.path.join(out, 'bad_500.txt')) + + # now it succeeds: the marker must go, otherwise a later run with + # --skip-errors would keep skipping an already processed document + self._run(client, inp, out, failing=(), force=True) + assert os.path.isfile(os.path.join(out, 'bad.grobid.tei.xml')) + assert not os.path.exists(os.path.join(out, 'bad_500.txt')) + + def test_new_error_replaces_marker_of_previous_status(self): + client = self._client() + with tempfile.TemporaryDirectory() as d: + inp = os.path.join(d, 'in') + out = os.path.join(d, 'out') + os.makedirs(inp) + self._make_pdfs(inp, ['bad.pdf']) + + self._run(client, inp, out, failing={'bad.pdf'}, force=True) + assert os.path.isfile(os.path.join(out, 'bad_500.txt')) + + def fake_post(url, files=None, data=None, headers=None, timeout=None): + resp = Mock() + resp.text = 'timeout' + return (resp, 408) + + with patch.object(GrobidClient, 'post', side_effect=fake_post): + client.process('processFulltextDocument', inp, output=out, force=True) + + assert os.path.isfile(os.path.join(out, 'bad_408.txt')) + assert not os.path.exists(os.path.join(out, 'bad_500.txt')) + + def test_skip_errors_counts_as_skipped(self): + client = self._client() + with tempfile.TemporaryDirectory() as d: + inp = os.path.join(d, 'in') + out = os.path.join(d, 'out') + os.makedirs(inp) + self._make_pdfs(inp, ['bad.pdf']) + self._run(client, inp, out, failing={'bad.pdf'}, force=True) + + processed, errors, skipped = client.process_batch( + 'processFulltextDocument', [os.path.join(inp, 'bad.pdf')], inp, out, + n=1, generate_ids=False, consolidate_header=False, + consolidate_citations=False, include_raw_citations=False, + include_raw_affiliations=False, tei_coordinates=False, + segment_sentences=False, force=False, skip_errors=True + ) + assert (processed, errors, skipped) == (0, 0, 1)