epo-processor

epo-processor

European Patent Office bulk data is useful and awkward at the same time. The BDDS product catalogue is large, each archive nests more archives, and the payloads that matter sit as XML exchange documents deep inside that tree. A naive download-and-unpack cycle fills disks and dies halfway through a multi-day run.

epo-processor is a streaming ETL for that problem. It pulls EPO archives from the EPO BDDS API, walks nested tar / gzip / zip containers without materialising whole trees, parses exchange documents, and writes Parquet. A second lane streams the HUPD all-years tarball to disk. A third lane joins the two worlds: citation overlap between EPO records and a local HUPD extraction, optionally as a linked Parquet dataset for downstream models, search indexes, or review tools.

What you get

Three commands share one pipeline shape.

process turns EPO BDDS product archives into Parquet. Archives are listed from the product catalogue, opened over HTTP (or replayed from a local directory), unwrapped in stream, and parsed into patent rows. An optional bbolt checkpoint records finished archives so a killed run continues instead of starting over.

process-hupd materialises the HUPD Hugging Face tarball. Same opener and walk machinery, but extraction and sink are no-ops: the goal is files on disk for later analysis, not Parquet rows.

analyze builds the bridge. It indexes HUPD identifiers (Feather metadata when you have it, otherwise a parallel JSON walk), streams the EPO Parquet or CSV once, and reports citation overlap. Pass --dataset to emit a linked Parquet where each retained EPO patent carries the HUPD paths and citation categories that matched.

Default behaviour is fully streaming: raw archives and extracted entries are not kept unless you opt in. Structured log/slog logs and end-of-run summary tables make long jobs observable without drowning the terminal.

Quick start

cd epo-processor
devenv shell

epo-build
epo-process                                      # config/config.yaml
epo-process-hupd data/hupd                       # optional HUPD lane
epo-analyze data/hupd ./data.parquet \
  --dataset data/epo_hupd_dataset.parquet

Installation

devenv (recommended)

The project ships a Nix-backed devenv shell with Go tooling, git-cliff, and container helpers. Enter the shell once, then use the epo-* scripts or Make.

# https://devenv.sh/getting-started/
curl -L https://get.devenv.sh | bash

cd epo-processor
devenv shell
make build                    # writes bin/epo-processor

Without devenv

git clone https://github.com/Qubut/epo-processor.git
cd epo-processor
go build -o bin/epo-processor ./cmd/epo_processor
./bin/epo-processor --help

Install

go install github.com/Qubut/epo-processor/cmd/epo_processor@latest

How the pipeline thinks

Every subcommand composes the same stages. Only the concrete implementations change.

Stages talk over typed streams from destel/rill, with backpressure and context.Context cancellation. Sink writes are serial; archive and extractor concurrency are the knobs that saturate cores and network.

Retention is a janitor concern, not buried inside openers. Flip pipeline.keep_archive or pipeline.keep_extracted when you need the raw bytes or XML on disk for debugging; leave both false for production streaming runs.

Resumability is a decorator on the source. Set pipeline.checkpoint_db to a bbolt path and completed archive IDs persist across process restarts. Leave it empty for a no-op checkpointer with identical control flow and no resume state.

Commands

Global flags: --config, --log-level / -l, --log-dir.

process

Downloads every archive in the configured BDDS product, unwraps nested containers, parses exchange documents, and batches rows into Parquet. Use a checkpoint when the run may span hours or days.

epo-process
# or:
epo-run process --config config/config.yaml -o ./data.parquet

Useful flags: -o / --out, -c / --concurrency, --extract-concurrency, --parser-concurrency, --batch-size, --batch-timeout, --row-group-size, --spool-dir, --local-dir, --keep-archive, --archive-dir, --keep-extracted, --extracted-dir, --checkpoint, --reset, --base-url, --timeout, --max-retries, --product-id, --verify-sha1.

process-hupd

Streams the HUPD all-years tarball and writes unwrapped entries under <out-dir>. Pass --keep-archive if you also want the raw .tar retained.

epo-process-hupd data/hupd
# or:
epo-run process-hupd data/hupd --keep-archive --archive-dir data/archives

Useful flags: -u / --url, --filename, --keep-archive, --archive-dir, --spool-dir, --extract-concurrency, --max-retries.

analyze

Indexes HUPD, streams EPO once, and reports overlap. Positional arguments are order-flexible: the directory is HUPD, the file is EPO. With a single argument, that path is the HUPD directory and EPO defaults to the pipeline output path from config (pipeline.output_parquet).

epo-analyze /data/hupd/extracted ./data.parquet \
  --dataset data/epo_hupd_dataset.parquet \
  -m /data/hupd/hupd_metadata.feather \
  -o report.json

Useful flags: -m / --meta, -o / --out, --dataset, --min-collisions, -w / --workers.

Configuration

Defaults live in config/config.yaml. Viper merges file, environment, and flags. Prefer editing the YAML for stable runs; use flags for one-off overrides.

log:
  log_level: info
  log_dir: logs

server:
  base_url: "https://publication-bdds.apps.epo.org/bdds/bdds-bff-service/prod/api/public"
  product_id: 3
  max_retries: 2
  timeout: 30s
  verify_sha1: false

hupd:
  url: "https://huggingface.co/datasets/HUPD/hupd/resolve/main/data/all-years.tar"
  filename: data/hupd_all-years.tar

analyze:
  hupd_meta_url: "https://huggingface.co/datasets/HUPD/hupd/resolve/main/hupd_metadata_2022-02-22.feather"

pipeline:
  archive_concurrency: 16
  extractor_concurrency: 0      # 0 = auto from CPU and memory budget
  parser_concurrency: 2
  memory_budget_gb: 32
  per_entry_estimate_mb: 256
  batch_size: 100000
  batch_timeout: 2s
  output_parquet: "./data.parquet"
  row_group_size: 500000
  spool_dir: ""
  use_local_dir: ""
  keep_archive: false
  archive_dir: "data/archives"
  keep_extracted: false
  extracted_dir: "data/xml"
  checkpoint_db: "data/.epo-state.db"

extractor_concurrency: 0 asks the planner to size in-flight XML readers from runtime.NumCPU() clamped by memory_budget_gb and per_entry_estimate_mb. Raise archive_concurrency if the network is idle while extractors wait; raise the memory budget if auto sizing undershoots a large machine.

Development

devenv shell

Scripts on $PATH (from devenv.nix):

Makefile targets:

make all runs tidy, lint, test, then build. Default make prints help.

API documentation

Package docs come from Go comments and examples. After a version tag is published they appear on pkg.go.dev:

https://pkg.go.dev/github.com/Qubut/epo-processor

go doc ./...
go doc github.com/Qubut/epo-processor/cmd

Container image

The prod image is a static binary plus CA certificates, built with devenv and nix2container. There is no Dockerfile.

devenv container build prod
devenv container run prod -- --help

docker pull ghcr.io/qubut/epo-processor:latest

Images publish from container.yml on main and v* tags.

Changelog and releases

Commits follow Conventional Commits. git-cliff regenerates CHANGELOG.md.

make changelog

Pushing a v* tag runs GoReleaser, refreshes CHANGELOG.md on main, and publishes the container image.

Contributing

1. Fork and branch.
2. Enter devenv shell.
3. Ensure make all passes.
4. Open a PR that states what changed and why.

Commit style: <scope>: <imperative> <what> (example: pipeline: tee kept entries).

License

MIT. See LICENSE.

Related