This repository contains scripts for converting scholarly bibliographic metadata from various data sources into the format accepted by OpenCitations Meta. The data sources currently supported are:
- About The Project
- Software Components
- ID Validation Process
- How to Run the Software
- How to Extend the Software
- License
- Contacts
- Acknowledgements
- References
The main function of this software is to perform a metadata crosswalk between the data sources providing bibliographic and citation data and the OpenCitations Data Model (OCDM). At the same time, the software also handles the normalization and validation of the identifiers provided by the data source in all cases where it is not also the registration agency for those identifiers. The software generates two main output datasets, based on the data provided by a specific source:
- Bibliographic entities CSV tables; which are meant to be used as input for the META software, an OpenCitations tool and database for managing bibliographic entities' metadata. Example:
| id | title | author | pub_date | venue | volume | issue | page | type | publisher | editor |
|---|---|---|---|---|---|---|---|---|---|---|
| doi:10.9799/ksfan.2012.25.1.069 | Nonthermal Sterilization and Shelf-life Extension of Seafood Products by Intense Pulsed Light Treatment | Cheigh, Chan-Ick [orcid:0000-0002-6227-4053]; Mun, Ji-Hye [orcid:0000-0002-6227-4053]; Chung, Myong-Soo | 2012-3-31 | The Korean Journal of Food And Nutrition [issn:1225-4339] | 25 | 1 | 69-76 | journal article | The Korean Society of Food and Nutrition [crossref:4768] |
- AnyID-to-Any-ID citations CSV tables; which will be used as input for the INDEX software, an OpenCitations tool and database for producing and managing citations between bibliographic entities identified by OMIDs (internal and unique identifiers assigned by OpenCitations). Example:
| citing | cited |
|---|---|
| doi:10.11426/nagare1970.2.4_1 | doi:10.1295/kobunshi.16.921 |
In practice, the outputs generated by oc_ds_converter are used in subsequent steps of the data ingestion process within the OpenCitations infrastructure. Specifically, the metadata tables are used as input for META. The software assigns an OMID identifier to new entities and propagates an existing OMID to the entities already present in the OpenCitations databases, thereby deduplicating identical entities ingested from different data sources.
Subsequently, the INDEX software, responsible for producing citation data compliant with the OpenCitations data model, takes as input the anyID-to-anyID citation tables produced by the oc_ds_converter software. It queries the META database to retrieve the OMID associated with each entity's identifier and produces OMID-to-OMID citations in various formats (RDF, SCHOLIX, CSV) as output.
Here, a diagram of the OpenCitations ingestion workflow:
The software is built upon three fundamental components, each addressing specific needs:
- Metadata Crosswalk (oc_ds_converter)
- Identifier Validation (oc_ds_converter/oc_idmanager)
- Data Storage Management (oc_ds_converter/oc_idmanager/oc_data_storage)
Each identifier schema has its own class (e.g.: PMIDManager(IdentifierManager), defined in oc_ds_converter/oc_idmanager/pmid.py), instantiated according to the model provided by the abstract class IdentifierManager(metaclass=ABCMeta), defined in oc_ds_converter/oc_idmanager/base.py.
Each class provides methods for:
- normalising the id string
- checking the correctness of the id syntax
- verifying its existence using specific API services (if available)
- In-memory storage (default): Uses a simple in-memory dictionary. Data is lost when the process ends. Suitable for single-threaded processing.
- SQLite storage: Persistent file-based storage. Suitable for single-threaded processing with persistence needs.
- Redis storage: Persistent networked storage. Required for multiprocessing (
--max_workers > 1).
The storage backend can be selected via command-line arguments:
- Default (no flags): in-memory storage
-s path/to/file.db: SQLite storage-s path/to/file.json: in-memory storage with JSON persistence-ror--use-redis: Redis storage (required for multiprocessing)
The temporary storage manager used while processing a data chunk is a simple in-memory dictionary wrapper (class BatchManager, defined in oc_ds_converter/oc_idmanager/oc_data_storage/batch_manager.py). This approach batches writes to the main storage for better performance. If the process stops mid-chunk, the data in BatchManager is lost and the chunk is reprocessed from the beginning on restart. Since validation results are idempotent, reprocessing simply overwrites the same values.
Subsequently, we perform another full iteration, validating all identifiers not registered by the data source itself.
Note that input datasets are typically composed of multiple files. Each file is processed independently, and a cache file tracks which files have been completed. During file processing, validation data is temporarily stored in BatchManager (see oc_ds_converter/oc_idmanager/oc_data_storage/batch_manager.py), a simple in-memory dictionary. When the file processing completes and the CSV output tables are produced, all accumulated data is transferred to the main storage in a single batch operation, reducing overhead compared to individual writes. The file is then marked as completed in the cache. If the process is interrupted mid-file, the file is not in the cache and will be reprocessed from the beginning on restart. However, IDs already stored in the main storage from previous operations do not require new API calls.
For each encountered identifier to be validated, an ordered list of checks should be performed, stopping as soon as the validity value can be assessed:
- Search for the identifier in the batch manager, containing data concerning the current data chunk;
- Search for the identifier in the main storage manager, containing data concerning the whole dataset;
- Search for the identifier in the OpenCitations databases, containing data of all the datasets ever ingested in OpenCitations.
- Use ID-schema specific API services to retrieve the validity information of the ID.
To produce the citations and metadata CSV output from a data source, it is possible to execute its specific process by selecting the correct source from oc_ds_converter/run/ directory. For example, the oc_ds_converter process for JaLC data source can be launched as follows:
# Single-threaded processing (default, in-memory storage)
python oc_ds_converter/run/jalc_process.py -ja /Volumes/my_disk/JALC_INPUT -out /Volumes/my_disk/JALC_OUTPUT
# Multi-threaded processing (requires Redis)
python oc_ds_converter/run/jalc_process.py -ja /Volumes/my_disk/JALC_INPUT -out /Volumes/my_disk/JALC_OUTPUT -ca /Volumes/my_disk/JOCI_CACHE.json -r -m 3This command launches a process of data conversion from the input data dump (located at /Volumes/my_disk/JALC_INPUT) into metadata CSV tables (stored at /Volumes/my_disk/JALC_OUTPUT) and citation CSV tables (stored in a directory automatically generated at /Volumes/my_disk/JALC_OUTPUT_citations). When using -r (Redis) and -m 3, the process uses up to 3 workers for parallelization. While the process is being executed, a cache file at /Volumes/my_disk/JOCI_CACHE.json is created and updated.
More in detail, each data source run script has a set of arguments that can be adapted to meet the peculiarities of the dataset. However, all the sources should accept a similar list of arguments:
- '--config': The path to a configuration file, where the other arguments can be declared;
- '--input_location': The path to the input data;
- '--output_location': The path to the output directory where the metadata CSV files will be stored. From the name of this directory, the name of the directory where to store the citation CSV files will be derived automatically.
- '--publishers': The path to an optional support CSV file containing additional information about publishers, their crossref members and the DOI prefix they are associated with (id, name, prefix), used to enrich the metadata.
- '--orcid': The path to an optional support table mapping DOIs to ORCIDs of the publications' authors, used to enrich the metadata.
- '--wanted': The path to an optional CSV filepath containing a list of DOIs to process.
- '--cache': The cache file path, that will be automatically deleted at the end of the process.
- '--storage_path': Path for ID validation storage. Use
.dbextension for SQLite or.jsonfor in-memory JSON storage. If not specified, uses in-memory storage. - '--use-redis': Use Redis for DOI-ORCID index and publishers lookup. Required for multiprocessing. By default, in-memory storage is used.
- '--testing': The parameter to define whether or not the script is to be run in testing mode. When testing mode is enabled, a fake in-memory Redis instance is used instead of a real Redis server.
- '--max_workers': The integer number of workers used to run the process in parallel executions. Requires
--use-redisto be enabled.
In order to manage a new data source, two main software components need to be developed:
- a script for reading the data source, extract the bibliographic entities' metadata, and produce the output tables;
- a script for reshaping the metadata of each bibliographic entity according to the OpenCitation data model.
In addition to that, if the data source uses persistent identifiers not managed by OpenCitations yet, a new identifier manager should be developed too.
For each new data source, a python file should be added to the directory oc_ds_converter/run/. The file should be named after the data source, and perform the following tasks:
- decopress and read the source dataset;
- manage the identifiers' validation process;
- extract from the source data a data structure representing each bibliographic resource;
- call a source-specific metadata crosswalk method to convert this data structure into an OCDM-compliant dictionary representing the bibliographic resource, to be stored as a CSV row;
- produce the output tables (citations and metadata)
All source represents bibliographic records according to a specific data model, which has to be mapped into OCDM. To do so, we implement a source-specific child class of the class RaProcessor (defined in oc_ds_converter/ra_processor.py) for each new data source. The main method of all RaProcessor children classes is csv_creator, which is aimed at producing a row for an OpenCitations metadata table from a data structure representing a bibliographic entry according the source data model. As an example, see OpenaireProcessing(RaProcessor) class (in oc_ds_converter/openaire/openaire_processing.py).
For adding a new ID Manager:
- create a python file at
oc_ds_converter/oc_idmanager, named after the id schema, e.g.oc_ds_converter/oc_idmanager/viaf.py. - create a new class as an instance of the abstract class
IdentifierManager(defined inoc_ds_converter/oc_idmanager/base.py), e.g.:ViafManager(IdentifierManager), thus following the provided template. In particular: - define all the id-schema specific required methods, i.e.:
syntax_ok, to check whether the ID is compliant to its own schema syntax,exists, to check the ID's existence using the ID-specific API,normalise, to normalise the identifier string (for example by removing unexpected character and turing the uppercase into lowercase characters), andis_valid, for assessing the overall validity of the identifier. - if possible, add additional ID-schema specific methods. For example, some ID schemas (such as ORCID and ISSN) are formed by following a specific check-digit mechanism, which provides a further control system to verify the ID validity: in these cases, it is possible to add also a
check_digitmethod.
For adding a new type of Storage Manager, i.e. relying on another storage system:
- create a python file at
oc_ds_converter/oc_idmanager/oc_data_storagenamed after the storage system, e.g.:oc_ds_converter/oc_idmanager/oc_data_storage/redis_manager.py. - create a new class as an instance of the abstract class
StorageManager(defined inoc_ds_converter/oc_idmanager/oc_data_storage/storage_manager.py), e.g.:RedisStorageManager(StorageManager), thus following the provided template. In particular: - define all the storage-type-specific required methods, i.e.:
set_value, to add a single key-value pair to the storage,set_multi_value, to store a list of key-value tuple pairs all at once,get_value, to retrieve the value associated to a specific key,del_value, to delete a key-value pair,delete_storage, to delete all the data previously saved in the storage system, andget_all_keys, to retrieve the list of all the keys in the storage.
The repository is managed with uv. To install dependencies:
uv sync
To add a package as a dependency to the project:
uv add <package>
To run all tests:
uv run pytest
To run specific tests:
uv run python -m unittest discover -s test -p "*.py"
Distributed under the ISC License. See LICENSE for more information.
- Arianna Moretti - @ariannamorettj - arianna.moretti4@unibo.it
- Arcangelo Massari - @arcangelo7 - arcangelo.massari@unibo.it
- Elia Rizzetto - @eliarizzetto - elia.rizzetto@studio.unibo.it
- Marta Soricetti - @martasoricetti - marta.soricetti@studio.unibo.it
Project Link: https://github.com/opencitations/oc_ds_converter
This project has been developed under the supervision of Prof. Silvio Peroni.
- Silvio Peroni - @essepuntato - silvio.peroni@unibo.it
- The OpenCitations Data Model
- OpenCitations Meta
- Metadata crosswalk for citation data production in OpenCitations
Files changed: oc_ds_converter/run/datacite_process.py, test/datacite_process_test.py
Problem. The function read_json() had an undeclared and irreversible side effect: when a file raised a JSONDecodeError during parsing, it physically moved the source file from the input directory to a _bad/ subdirectory inside the output folder. This behavior was invisible from the function signature and contradicted the expectation that a function named "read" is non-destructive.
Critical point in the previous code. The signature was read_json(json_path, bad_dir, preview_chars). Internally, on JSONDecodeError, the function called shutil.move(json_path, bad_dir) before returning. This meant that merely calling read_json() on a malformed file permanently altered the filesystem state. Additionally, the function only caught JSONDecodeError — other I/O failures such as PermissionError or a genuine FileNotFoundError (e.g. a path typo in the caller) propagated as unhandled exceptions and would crash the entire process.
Proposed update. The bad_dir parameter and all file-moving logic were removed from read_json(). The function now logs the error and returns None, leaving the source file in place. A broad except Exception clause was added to absorb non-JSON I/O errors:
def read_json(json_path, preview_chars: int = 100):
try:
...
except JSONDecodeError as e:
print(f"[JSON ERROR] file={json_path}: {e}\n preview: {preview}")
return None
except Exception as e:
print(f"[READ ERROR] file={json_path}: {e}")
return NoneParts of the code affected. oc_ds_converter/run/datacite_process.py — read_json() function (signature and body). All call sites of read_json() in preprocess() updated to drop the removed bad_dir argument.
How the behavior changed. read_json() is now a pure read function with no filesystem side effects. A malformed or unreadable file is logged and skipped; the file itself is not touched. All I/O errors are handled gracefully rather than crashing the process.
Problem. In single-threaded mode, preprocess() maintained a local list bad_list to track files that had failed in Pass 1. This list was used to skip those same files in Pass 2. The bad_list was not a deliberate policy choice — it was a structural workaround made necessary by the file-moving side effect in read_json().
Critical point in the previous code. After read_json() moved a malformed file in Pass 1, the file no longer existed at its original path. When Pass 2 iterated over all_input_json (the original full file list), it called read_json() on the now-missing path. Since read_json() only caught JSONDecodeError (not FileNotFoundError), this raised an unhandled exception in the Pass 2 loop and could crash the process or silently skip all remaining files. bad_list existed solely to prevent this crash by skipping the already-moved files before read_json() was called on them.
Proposed update. With read_json() no longer moving any files (see §1.1), the root cause of the crash was eliminated. bad_list and the bad_dir variable were removed from preprocess(). Pass 2 now iterates over all_input_json unconditionally, exactly like Pass 1. Files that read_json() cannot parse are skipped via the already-present if chunk: guard — no separate tracking list is needed.
Parts of the code affected. oc_ds_converter/run/datacite_process.py — preprocess() function, single-threaded path: removed bad_list initialization, bad_list.append() calls, and the if json_file not in bad_list guard in the Pass 2 loop.
How the behavior changed. The single-threaded pass structure is now symmetric: both Pass 1 and Pass 2 iterate over the same file list with the same skip logic. There are no tracking lists that could fall out of sync with the actual file list.
Problem. In multiprocessing mode, the bad_list workaround from the single-threaded path (see §1.2) was never implemented. As a result, when files were moved by read_json() in Pass 1, Pass 2 called read_json() on the now-missing paths with no protection. The resulting FileNotFoundError was unhandled, silently breaking Pass 2 for every file that had failed in Pass 1.
Critical point in the previous code. The multiprocessing path in preprocess() iterated over all_input_json for both passes without updating the list after Pass 1. The assumption was that read_json() was safe to call on any path in the list; once that assumption was violated by the file-move side effect, Pass 2 raised uncaught exceptions. Because multiprocessing errors can surface in non-obvious ways (depending on how the executor handles exceptions), this failure mode could go unnoticed: Pass 2 might produce no output for affected files without printing a clear error.
Proposed update. As with the single-threaded path, the fix is entirely in read_json() (see §1.1). With files no longer moved, all_input_json remains valid for the duration of both passes. No changes to the multiprocessing pass loop logic were necessary beyond dropping the removed bad_dir argument from read_json() call sites.
Parts of the code affected. oc_ds_converter/run/datacite_process.py — preprocess() function, multiprocessing path: read_json() call sites updated to remove bad_dir argument.
How the behavior changed. Multiprocessing mode now handles malformed files consistently with single-threaded mode. A file that fails in Pass 1 is skipped (returns None, if chunk: guard fires), remains in the input directory, and is automatically eligible for retry in Pass 2 of the same run and in any subsequent run.
Problem. The DataCite file-error policy (move to _bad/) was at odds with the Crossref policy (skip and leave in place), making behavior inconsistent across two processors that are otherwise structurally aligned.
Critical point in the previous code. In crossref_process.py, a file that fails to parse returns False from its read function, is logged, and is left untouched. It will be retried on the next run. In datacite_process.py, the same type of failure caused the file to be permanently moved to a _bad/ directory. This made the DataCite behavior irreversible: a temporarily malformed file (e.g. a network-truncated download that was later re-downloaded correctly) would need to be manually moved back before it could be reprocessed.
Proposed update. DataCite now follows the same policy as Crossref: a file that cannot be parsed is skipped for the current run, is not added to the cache (so it will be retried on the next run), and is left at its original path. No _bad/ directory is created.
Parts of the code affected. oc_ds_converter/run/datacite_process.py — read_json() (file-moving logic removed) and preprocess() (bad_dir variable and _bad/ directory creation removed). The _bad/ directory is no longer created at any point.
How the behavior changed. Failed files are recoverable without manual intervention. If a file was malformed due to a temporary condition (truncated download, filesystem issue), fixing the source file is sufficient for it to be processed on the next run. The behavior is now consistent with crossref_process.py.
test/datacite_process_test.py — test_malformed_file_is_skipped_not_moved
Creates a malformed .json file in a temporary input directory, runs preprocess(), and asserts:
- The source file still exists at its original path (not moved)
- No
_bad/directory was created in the output folder
Files involved: oc_ds_converter/run/datacite_process.py (and all other run/ scripts)
2.1 — Cross-platform incompatibility: implicit dependency on the fork start method via pebble.ProcessPool
Problem. The DataCite multiprocessing implementation could not run on Windows and macOS (in newer Python versions where spawn is the default), because it implicitly depended on the fork process start method, which is only available on Linux.
Critical point in the previous code. datacite_process.py used pebble.ProcessPool (a third-party library) to manage worker processes. pebble.ProcessPool does not accept an explicit mp_context argument, so the process start method was inherited from the operating system default. On Linux the default is fork (inherits parent memory, no need to re-import modules). On Windows the only available method is spawn (starts a fresh interpreter, requires all worker functions to be importable as top-level module names). On macOS, Python 3.8+ changed the default from fork to spawn. A workaround had been applied at some point by removing the mp_context argument entirely, which did not resolve the underlying incompatibility — it just removed the explicit evidence of it. Worker functions and their arguments must be picklable for spawn to work; pebble.ProcessPool without mp_context provided no guarantee of this.
Proposed update. pebble.ProcessPool was replaced with concurrent.futures.ProcessPoolExecutor from the Python standard library, with an explicit mp_context=get_context('spawn') argument:
from multiprocessing import get_context
from concurrent.futures import ProcessPoolExecutor
with ProcessPoolExecutor(max_workers=max_workers, mp_context=get_context('spawn')) as executor:
futures = [executor.submit(process_chunk, ...) for ...]
for fut in futures:
fut.result()The spawn start method is supported on Linux, macOS, and Windows. Using get_context('spawn') makes the choice explicit and reproducible regardless of OS defaults.
Parts of the code affected. oc_ds_converter/run/datacite_process.py — preprocess() multiprocessing block: pebble.ProcessPool replaced with ProcessPoolExecutor. Import block: pebble removed, from multiprocessing import get_context and from concurrent.futures import ProcessPoolExecutor added. This aligns datacite_process.py with the approach already in use in crossref_process.py, openaire_process.py, jalc_process.py, and (now) pubmed_process.py.
How the behavior changed. The DataCite multiprocessing path now runs correctly on Windows and macOS without OS-specific workarounds. The explicit mp_context=get_context('spawn') makes the start method part of the code contract rather than an implicit OS dependency.
test/datacite_process_test.py — test_multiprocessing_spawn_context_produces_output
Runs preprocess() with max_workers=2, redis_storage_manager=True, and testing=True (FakeRedis). Asserts that the multiprocessing path completes without errors and produces the same citation and entity counts as the single-threaded path (19 citations, 22 entities).
Files changed: oc_ds_converter/run/pubmed_process.py, test/pubmed_process_test.py
Files deleted: oc_ds_converter/run/pubmed_process_new.py
Problem. The original pubmed_process.py processed PubMed CSV input in a single sequential loop with no parallelism. This was a significant performance bottleneck because PubmedProcessing.csv_creator() makes external API calls (NIH API for journal metadata, publisher prefix resolution) for each entity; in a single-threaded loop, these calls are strictly sequential.
Critical point in the previous code. The cache used a plain-text integer counter (a file containing a number such as 3, meaning "3 files have been completed"). This format is incompatible with the JSON chunk-range format used by all other processors in the pipeline ({"citing": [[0, 6], [7, 13], ...]}). The integer counter could only track which whole files had been completed; it could not record progress within a partially-processed file. If the process was interrupted mid-file, the entire file would be reprocessed from row 0 on the next run.
Proposed update. pubmed_process.py was fully rewritten to adopt the same chunk-based architecture used in datacite_process.py and jalc_process.py. Three chunk distribution functions — find_missing_chuncks, new_chunks_distribution, and assign_chunks — implement a state machine that divides a CSV file into row-range slices [start_row, end_row] and records each completed slice in a per-file JSON cache. A ProcessPoolExecutor with mp_context=get_context('spawn') (cross-platform, consistent with DataCite/JaLC) dispatches chunks to worker processes when max_workers > 1. In single-threaded mode (max_workers=1, the default), chunks are processed sequentially in the main process. Progress is displayed via create_progress() / advance_progress() from oc_ds_converter.lib.console (Rich-based), replacing the previous tqdm bar.
Parts of the code affected. oc_ds_converter/run/pubmed_process.py (entire file rewritten). New functions: find_missing_chuncks, new_chunks_distribution, assign_chunks, _count_rows, _read_row_range, _write_csv, _mark_chunk_done, process_chunk, preprocess. New --max_workers argument added to the __main__ argument parser.
How the behavior changed. The process can now be parallelized using multiple workers. Progress within a partially-processed file is preserved: if the run is interrupted, only the incomplete chunks are reprocessed on restart, not the entire file. Per-file cache files ({base}_cache.json) are deleted automatically after successful file completion, consistent with DataCite behavior.
Problem. A second file, pubmed_process_new.py, existed in oc_ds_converter/run/ as an incomplete attempt to align the PubMed script with the JaLC/DataCite structure. Its presence was a risk: it was discoverable and runnable by name, but would crash immediately on any real PubMed input.
Critical point in the previous code. pubmed_process_new.py was a near-direct copy of jalc_process.py. Despite being meant for PubMed, it imported JalcProcessing instead of PubmedProcessing and made no use of PubMed-specific logic (journals_filepath, orcid_index, publishers_filepath_pubmed, exclude_existing). The file contained three correctly implemented and tested chunk distribution functions (find_missing_chuncks, new_chunks_distribution, assign_chunks) that were worth preserving, but all other code was wrong. The test file (test/pubmed_process_test.py) imported from pubmed_process_new via a wildcard import (from oc_ds_converter.run.pubmed_process_new import *), creating a dependency on a broken file.
Proposed update. pubmed_process_new.py was deleted. The three correct chunk distribution functions were migrated verbatim into the new pubmed_process.py. The test file import was updated to an explicit named import from pubmed_process:
from oc_ds_converter.run.pubmed_process import (
assign_chunks, find_missing_chuncks, new_chunks_distribution, preprocess,
)Parts of the code affected. oc_ds_converter/run/pubmed_process_new.py (deleted). test/pubmed_process_test.py (import statement updated).
How the behavior changed. There is now a single authoritative PubMed run script. The three chunk tests (test_find_missing_chuncks, test_new_chunks_distribution, test_assign_chunks) continue to pass unchanged, since the functions are reproduced exactly.
Problem. The initial rewrite of pubmed_process.py followed the DataCite/JaLC two-pass design, which produces both a bibliographic entity table (for META) and a citation table (for INDEX). For PubMed, generating a citation table is both unnecessary and redundant.
Critical point in the previous code. DataCite and JaLC require a converter-produced citation table because they are not the sole registration authority for their identifiers and the citation data needs to be paired with validated IDs. PubMed is different: the NIH is the sole registration authority for PMIDs, so PMID validation is trivially guaranteed. More importantly, the NIH already distributes a pre-built citing/cited table in exactly the format required by INDEX. Producing a second citation table from pubmed_process.py would duplicate existing data without adding value, and would add a second processing pass over the entire dataset.
Proposed update. The design was changed to a single-pass entity extraction. preprocess() runs only one pass over each CSV file, producing only the bibliographic entity table for META. No {csv_dir}_citations/ directory is created. The assign_chunks JSON cache uses only the "citing" key, which in this context tracks the entity extraction pass (the naming is inherited from the shared chunk infrastructure and reflects the pass semantics of other processors).
Parts of the code affected. oc_ds_converter/run/pubmed_process.py — the preprocess() function has no second loop and no citations output path. test/pubmed_process_test.py — test_preprocess_creates_entities explicitly asserts that no _citations/ directory is created.
How the behavior changed. The process is faster (one pass instead of two) and produces no spurious output. The architecture correctly reflects the source-specific constraint: PubMed citation data comes from the NIH directly, not from this converter.
Problem. The entity row filter — intended to pass only entities involved in at least one citation to META — was silently not working for one of its two conditions.
Critical point in the previous code. The filter logic was if r.get("references") or r.get("cited_by"). The cited_by column holds the PMID(s) of other articles that cite the current entity. If cited_by is populated, the entity is the target of a citation — it is cited by something else and its metadata is therefore needed by META. However, cited_by was absent from the usecols list passed to pandas.read_csv in _read_row_range(). Pandas silently ignores missing columns in usecols rather than raising an error, so the column was never loaded. The filter evaluated r.get("cited_by") as None for every row, making that branch of the condition permanently false. As a result, entities that appeared only in the cited_by field of other rows (i.e., entities that are cited but do not themselves cite anyone) were excluded from the output.
Proposed update. cited_by was added to _CSV_FILTER (the canonical column list used for both usecols and _CSV_DTYPE):
_CSV_FILTER = ["pmid", "doi", "title", "authors", "year", "journal",
"references", "cited_by"]
_CSV_DTYPE = {col: str for col in _CSV_FILTER}The filter if r.get("references") or r.get("cited_by") in process_chunk now correctly evaluates both columns.
Parts of the code affected. oc_ds_converter/run/pubmed_process.py — _CSV_FILTER constant and process_chunk row filter.
How the behavior changed. Entities that are cited by other entities in the dataset (i.e., cited_by is non-empty) are now correctly included in the META output. This restores the intended semantics of the filter: only entities involved in at least one citation — either as citing or cited party — are sent to META.
Problem. The new _write_csv() function, as initially written, would silently overwrite an existing output file if a second chunk happened to resolve to the same output path. This was a regression relative to the original pubmed_process.py, which included collision avoidance logic.
Critical point in the previous code. The output filename for a chunk is constructed as {base}_{file_num}.csv. In normal operation, file_num derives from start // interval + 1, which is unique per chunk within a single file. However, in edge cases (e.g., two runs with different interval values targeting the same csv_dir, or a restart that re-derives the same chunk number), two chunks could produce the same path. The new _write_csv() initially used a plain open(filepath, 'w'), which would truncate and overwrite any existing file at that path. Any entity data already written there would be permanently lost.
Proposed update. The collision avoidance logic was restored. If filepath already exists at write time, a datetime timestamp is appended before the .csv extension:
def _write_csv(rows: list[dict], filepath: str) -> None:
if not rows:
return
if os.path.exists(filepath):
dt = datetime.now().strftime("%d%m%Y_%H%M%S")
filepath = filepath[:-4] + "_" + dt + ".csv"
...from datetime import datetime was added to the imports of pubmed_process.py.
Parts of the code affected. oc_ds_converter/run/pubmed_process.py — _write_csv() function and import block.
How the behavior changed. No entity data can be lost due to a filename collision. Each chunk's output receives a unique path. This restores the minimum data-safety guarantee that existed in the original script and aligns with the general principle that output operations must never silently discard previously written data.
Problem. In max_workers > 1 mode, multiple worker processes may corrupt the publisher prefix cache and journal metadata cache through concurrent uncoordinated writes.
Critical point in the current code. PubmedProcessing maintains two JSON support files: publishers_filepath (publisher prefix cache) and journals_filepath (journal name/ISSN map). During csv_creator(), each worker reads from and writes to these files via save_updated_pref_publishers_map(). These read-modify-write operations are not protected by any file lock. When two workers run simultaneously, the standard race condition applies: Worker A reads the file, Worker B reads the file, Worker A writes its update, Worker B writes its update (overwriting A's changes). The update from Worker A is lost. Over a large dataset with many workers, this can produce a progressively desynchronized or partially corrupted support file.
Current status. No fix has been applied. Fixing the race condition would require modifications to PubmedProcessing.save_updated_pref_publishers_map() — adding a FileLock around its read-modify-write cycle — which is a change to the processing class, not the run script, and is outside the scope of this rewrite. In max_workers=1 mode (the default, and the only safe mode for now), the risk does not exist because operations are strictly sequential.
Parts of the code affected. oc_ds_converter/pubmed/pubmed_processing.py — save_updated_pref_publishers_map() (change not yet applied). oc_ds_converter/run/pubmed_process.py — process_chunk() (call site, change not yet applied).
How the behavior is affected. Running with max_workers > 1 may produce incomplete or inconsistent publisher and journal support files. Entity CSV output is not affected (each worker writes to a uniquely named file). The issue affects only the quality of the lookup caches that enrich publisher and journal metadata in subsequent runs. This limitation should be addressed before max_workers > 1 is used in production.
test/pubmed_process_test.py was updated as follows:
- Import updated: from
from oc_ds_converter.run.pubmed_process_new import *to explicit named imports frompubmed_process(see §3.2). - Three existing chunk tests unchanged:
test_find_missing_chuncks,test_new_chunks_distribution,test_assign_chunks— all pass without modification, since the functions are exact copies. - New test added:
test_preprocess_creates_entities— runspreprocess()withmax_workers=1,interval=5,testing=Trueon the short CSV fixture (test/pubmed_process/csv_files_short/). Asserts that at least one entity CSV file is created, that at least one entity row is present in the output, and that no_citations/directory is created (see §3.3). - Two tests commented out with explanatory notes:
test_preprocess_baseandtest_preprocess_interval_number— these tested the old single-threaded implementation with entity-count-based batching semantics.test_preprocess_baseassertions on content are compatible with the new implementation and can be re-enabled after verifying output column names.test_preprocess_interval_numberrelies on the old assumption thatintervalcontrols the number of entities per output file; in the new implementation,intervalcontrols the number of source CSV rows per chunk (not the number of entities in the output), so the expected entity counts per file would need to be recalculated before this test is re-enabled.
Planned interventions, ordered by priority:
Files: oc_ds_converter/openaire/openaire_processing.py, run/openaire_process.py
- Profile execution on a small sample to identify the bottleneck (~3 s/row)
- Check whether the slowdown is in external API calls, Redis access, or parsing
- Compare the processing loop with the one in
crossref_process.py - Implement the identified optimizations (e.g. batching, local cache for repeated lookups, reduction of external calls)
- Retest on the same 255-entry sample used on 2026-06-23 to measure the improvement