Harvesting for SCIGMA

Installation

1. Install the package, for example with uv:

uv pip install scigma_harvesting

2. Copy .env.example to .env.

cp .env.example .env

3. Fill in your local API keys and IDs.

4. Optional you may adapt config.yaml to your needs. It contains all configuration values.

Usage

For working examples, see:

• Minimal workflow – Basic harvest pipeline with Zotero/JSON/RIS export.
• Advanced raw queries – Using raw query parameters (base_raw, ddb_raw, stcv_raw) for source-specific control.

Configuration

Most functions require a config parameter that is loaded from a YAML configuration file:

from scigma_harvesting import load_config, harvest

config = load_config("./config.yml")
records = harvest(config=config, must=["Aristoteles"])

See Configuration below for details on the config file format.

Top-Level Functions

The scigma_harvesting package exposes the following functions:

• harvest(config, ...) – Main search function. Queries all sources (STCV, BASE, DDB), deduplicates results, returns list[HarvestRecord].
• harvest_all(config, ...) – Like harvest, but without deduplication.
• deduplicate(records, custom_filter=None) – Deduplicate HarvestRecords directly. Accepts optional custom_filter function for additional post-processing.
• to_zotero(config, records, ...) – Exports HarvestRecords to Zotero (via API).
• to_ris(records, path) – Exports HarvestRecords to a RIS file.
• to_json(records, path) – Exports HarvestRecords to a JSON file.
• activate_console_output(debug=False) – Configure console logging: INFO level by default, DEBUG level when debug=True.
• load_config(path) – Load configuration from a YAML file.

Source-specific search functions (all require their respective config):

• base_search(config, log_file, query, ...) – Search BASE API directly.
• ddb_search(config, log_file, query, ...) – Search Deutsche Digitale Bibliothek directly.
• stcv_search(config, log_file, ...) – Search STCV database directly.

Note: log_file must be passed as a keyword argument.

Data Model

The normalized data model HarvestRecord of the output is importable from the package and documented in src/scigma_harvesting/pipeline/models.py.

Source-Specific Functions and Raw Dumps

Each search function (base_search, ddb_search, stcv_search) accepts a log_file parameter to save raw API/database responses for debugging and reproducibility. This is especially useful for investigating query issues or preserving data snapshots. Note that log_file is required and must be passed as a keyword argument.

Additionally, stcv_search supports a sql_query parameter for direct SQL queries against the STCV SQLite database. When provided, it bypasses all other filter parameters and executes the raw SQL, which must return rows containing a cloi column.

BASE (scigma_harvesting.base)

from scigma_harvesting import load_config, base_search

config = load_config("./config.yml")

results = base_search(
    config=config.base,
    log_file=Path("dumps/base_response.jsonl"),
    query="dccreator: lossau",
)

Dump contents: JSON Lines file containing source, query, params, and raw_xml_file (path to separate .xml file with pretty-printed raw response). Each search call appends a new line.

Special behavior: Creates an additional files:

• <log_file>.xml – Raw XML response from BASE API (pretty-printed)

DDB (scigma_harvesting.ddb)

from scigma_harvesting import load_config, ddb_search

config = load_config("./config.yml")

results = ddb_search(
    config=config.ddb,
    log_file=Path("dumps/ddb_response.jsonl"),
    query="lossau",
)

Dump contents: JSON Lines file containing source, query, params, and raw_response (complete DDB API JSON response). Each search call appends a new line.

STCV (scigma_harvesting.stcv)

from scigma_harvesting import load_config, stcv_search

config = load_config("./config.yml")

# Standard search with filters
results = stcv_search(
    config=config.stcv,
    log_file=Path("dumps/stcv_response.jsonl"),
    standalone=True,
    must_terms=["Aristoteles"],
)

# Direct SQL query (must return rows with cloi column)
results = stcv_search(
    config=config.stcv,
    log_file=Path("dumps/stcv_sql_response.jsonl"),
    sql_query="SELECT cloi FROM title WHERE title_ti LIKE '%Aristoteles%'",
)

Documentation contents: JSON Lines file containing source, search parameters (standalone, must_terms, must_not_terms, year_from, year_to, sql_query), and results (parsed records). Each search call appends a new line.

Note: STCV uses a local SQLite database, so dumps contain the already-parsed results rather than raw API responses.

Configuration

The package uses a centralized YAML-based configuration system. All API endpoints, paths, and other settings are managed through a configuration file.

Default config file: config.yml (can be specified when loading)

# BASE API
base:
  url: "https://api.base-search.net/cgi-bin/BaseHttpSearchInterface.fcgi" # BASE API endpoint
  default_hits: 10 # Default results per page
  max_hits: 120 # Maximum results per request
  max_offset: 999 # Maximum offset for pagination

# DDB (Deutsche Digitale Bibliothek)
ddb:
  url: "https://api.deutsche-digitale-bibliothek.de/2/search/index/search/select" # DDB API endpoint
  # ddb_time_parser_cache: "./data/internal/ddb/ddb_timeparser_cache.tsv"  # Legacy, now hardcoded in ddb_cache.py

# STCV (Short Title Catalogue Vlaanderen)
stcv:
  url: "https://anet.be/opendata/stcv/stcv.sqlite.gz" # STCV API endpoint
  data_path: "./data/internal/stcv/latest/" # Local database path
  timeout: 30 # Download/processing timeout in seconds

# Documentation
documentation:
  console_debug: false # Enable console debug logging. Do not use debug in production!
  log_dir: "./data/internal/logs" # Directory for debug log files

# Zotero API (environment variables are resolved from .env)
zotero:
  chunk_size: 50 # Items per batch request
  chunk_delay: 2.0 # Delay between batches in seconds
  http:
    read_timeout: 60.0 # HTTP read timeout in seconds
    connect_timeout: 30.0 # HTTP connect timeout in seconds

API keys for BASE and Zotero are stored in your .env file and loaded separately (not resolved automatically by load_config()). You can also build the config programmatically:

from pathlib import Path
from scigma_harvesting import Config, BaseConfig, DdbConfig, StcvConfig, ZoteroConfig, ZotHttpConfig

config = Config(
    base=BaseConfig(url="...", default_hits=10, max_hits=120, max_offset=999),
    ddb=DdbConfig(url="...", ddb_time_parser_cache=Path("./data/internal/ddb/ddb_timeparser_cache.tsv")),
    stcv=StcvConfig(url="...", data_path=Path("./data/internal/stcv/latest/"), timeout=30),
    zotero=ZoteroConfig(chunk_size=50, chunk_delay=2.0, http=ZotHttpConfig(read_timeout=60.0, connect_timeout=30.0)),
    documentation=DocumentationConfig(console_debug=False, log_dir=Path("./data/internal/logs"))
)

Logging

The package uses Python's logging module. All modules (base, ddb, stcv, pipeline) log at DEBUG and INFO levels:

[!WARNING] When debug=True HTTP request URLs and headers are logged, which may include API keys. Never use debug=True in production or in environments where logs are shared or persisted.

• DEBUG: Raw API requests/responses (query parameters, full XML/JSON responses), individual record details, Zotero batch creation summaries.
• INFO: High-level progress (search start/end, result counts, harvest completion, Zotero chunk processing).

To enable detailed logging in your own code, use the provided helper:

from scigma_harvesting import activate_console_output

# Activate console logging (INFO level by default)
activate_console_output()

# Or for DEBUG level (very verbose, including API requests/responses):
activate_console_output(debug=True)

Troubleshooting

Common issues:

• BASE API returns no results: Verify API_KEY_BASE in .env. Check query syntax (Lucene/SOLR). BASE has a hard limit of ~1080 results per query.
• Zotero export fails: Verify API_KEY_ZOTERO and USER_ID_ZOTERO. Check collection key exists.
• DDB IIIF/PDF links not accessible: Many DDB resources require institutional authentication.

Debug mode: Enable detailed logging or run development tests for debugging:

from scigma_harvesting import activate_console_output
activate_console_output(debug=True)  # Enable DEBUG logging for all HTTP requests/responses

Or run dev tests to inspect individual components:

uv run pytest -m dev -v

Debug logging in integration tests: To enable DEBUG logging for integration tests (e.g., to inspect HTTP requests), use the --log-debug flag:

# Without debug (default: no secrets in logs)
pytest tests/integration/

# With debug (WARNING: may log API keys and other secrets!)
pytest --log-debug tests/integration/

Note: The --log-debug flag is disabled by default. When enabled, API keys may appear in log files (tests/artifacts/logs/test_e2e_debug.log). Never use --log-debug in CI or shared environments.

Project Structure

src/scigma_harvesting/
├── base/       # BASE API integration
├── ddb/        # Deutsche Digitale Bibliothek
├── stcv/       # Short Title Catalogue Vlaanderen
└── pipeline/   # Core harvest pipeline
    ├── core/   # Main pipeline module (harvest, harvest_all)
    ├── models  # HarvestRecord and Zotero export models
    └── export  # Export functions (to_zotero, to_ris, to_json)

IO Quality

The tests/integration/test_io_quality.py module contains integration tests that verify data integrity during I/O operations (searching, processing, saving, loading). For manual comparision dumps can be found in tests/artifacts/io-quality after running this test.

DDB

• API Access: The DDB search API is public and does not require an API key.
• MODS (MARC21) URLs: Generated via /items/{id}/source/record — publicly accessible.
• IIIF endpoints: Some IIIF manifests linked in DDB records may require institutional authentication tokens (not provided by this library).
• METS endpoints: Typically require authentication and are not publicly accessible.

Updating the Cache

The DDB uses a custom day-based timestamp system for dates. The DDB_TIMEPARSER_CACHE dictionary in scigma_harvesting.ddb.ddb_cache contains precomputed day timestamps for January 1st of each year.

If you need to update the DDB time parser cache (e.g., to extend it to cover more years):

1. Place your updated ddb_timeparser_cache.tsv file in data/internal/ddb/
2. Run the conversion script:

uv run python scripts/convert_ddb_cache.py data/internal/ddb/ddb_timeparser_cache.tsv src/scigma_harvesting/ddb/ddb_cache.py

This will generate a new ddb_cache.py file with the updated cache as a Python dictionary.

The TSV file format is simple: each line contains <year>\t<ddb_day_timestamp> where <ddb_day_timestamp is the time_stamp for the first day of the year.

Development

Git and Versioning

GitFlow

Merge non-squashing from dev to main. Fast-forwarding is not allowed in order to see the different releases on main as merge commit.

Versioning

Versioning is handled primarily via the version in pyproject.toml. To bump run

uv version --bump [beta|patch|minor|major]

Then commit and push.

In order to make a release from that, a git tag will trigger the CI:

# git tag
git tag -a $(uv version | awk '{print $2}') -m "Release $(uv version | awk '{print $2}')"
# git push tag
git push origin $(uv version | awk '{print $2}')

Styling

The code is formated by black formatter.

Tests

Run all tests (including dev tests, not recommended for production!) with:

uv run pytest -v

Test markers:

• -m prod – Production tests (safe for CI/CD, includes STCV database initialization).
• -m dev – Development/debug tests only (not suitable for production).

If errors occur, you can run individual tests or test modules for debugging:

# Run a specific test file
uv run pytest tests/unit/pipeline/test_pipeline.py -v

# Run a single test function
uv run pytest tests/unit/pipeline/test_pipeline.py::test_base_raw_passed_through_directly -v