Documentation
¶
Overview ¶
Package config loads and validates the YAML / environment-variable configuration for the streaming EPO/HUPD processor.
Subcommands ¶
Two streaming subcommands share one Config tree:
- process — EPO XML → Parquet
- process-hupd — HUPD .tar → disk
The legacy download / extract / parse chain has been removed; only the fields used by the streaming pipeline are recognised.
Loading Order ¶
- Built-in defaults (Config zero values + viper.SetDefault).
- YAML file (path supplied to [LoadConfig], typically config/config.yaml).
- Environment variables (uppercase, dot → underscore).
Validation ¶
Every [LoadConfig] call runs the struct through go-playground/validator and fails fast on missing required fields. Invariants beyond struct tags (e.g. spool dir writability) are checked in cmd at start-up.
Environment Variables ¶
All keys are bindable. Examples:
EPO_PROCESSOR_PIPELINE_CHECKPOINT_DB=/var/lib/epo/state.db EPO_PROCESSOR_TELEMETRY_ENABLED=true
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Analyze ¶
type Analyze struct {
// HUPDMetaURL is the URL of the HUPD metadata Feather file to download
// when it is not present at the path given by --hupd-meta.
// Default: the 2022-02-22 snapshot on HuggingFace.
HUPDMetaURL string `mapstructure:"hupd_meta_url"`
}
Analyze configures the `analyze` subcommand.
type Config ¶
type Config struct {
Log Log `mapstructure:"log" validate:"required"`
Server Server `mapstructure:"server"`
HUPD HUPD `mapstructure:"hupd"`
Analyze Analyze `mapstructure:"analyze"`
Pipeline Pipeline `mapstructure:"pipeline"`
}
Config is the root configuration object. Each subsection is a value type so it can be passed around freely; viper validates on Load().
func Load ¶
Load reads config from file/env/flags/defaults and validates it.
binds maps a CLI flag name to its dotted viper config key (e.g. "archive-concurrency" -> "pipeline.archive_concurrency"). Only flags listed in binds are bound to config; this keeps short, command-local flag names decoupled from the (stable) config-file/env key schema. Either argument may be nil/empty.
type HUPD ¶
type HUPD struct {
// URL is the full download URL of the HUPD .tar file.
URL string `mapstructure:"url"`
// Filename is used as the archive name in log messages and
// as the key in the checkpoint database.
Filename string `mapstructure:"filename"`
}
HUPD is the HuggingFace HUPD .tar source consumed by `process-hupd`.
type Log ¶
type Log struct {
// LogLevel controls the minimum severity emitted.
// Accepted values (case-insensitive): debug, info, warn, error.
// Default: info.
LogLevel string `mapstructure:"log_level" validate:"required,oneof=debug info warn error"`
// LogDir is the directory where the JSON log file is written.
// An empty string disables file logging (console logging is unaffected).
LogDir string `mapstructure:"log_dir"`
}
Log configures the slog logger (level and output directory).
type Pipeline ¶
type Pipeline struct {
// ArchiveConcurrency is the number of archives fetched and walked
// in parallel. Default: 4.
ArchiveConcurrency int `mapstructure:"archive_concurrency"`
// ExtractorConcurrency is the number of archive entries decoded in
// parallel (one sequential, CPU-bound XML reader each) — the knob that
// saturates cores. 0 (or negative) means auto: runtime.NumCPU() clamped
// by MemoryBudgetGB / PerEntryEstimateMB. Default: 0 (auto).
ExtractorConcurrency int `mapstructure:"extractor_concurrency"`
// ParserConcurrency is the number of goroutines that run the CPU-bound
// per-document parse in parallel within each entry. Since the reader is
// sequential, a small value (just enough to overlap parse with read) is
// sufficient. 0 (or negative) defaults to a small internal constant.
ParserConcurrency int `mapstructure:"parser_concurrency"`
// MemoryBudgetGB caps the RAM devoted to concurrent readers when
// ExtractorConcurrency is auto (0). Each in-flight reader buffers a
// decompressed XML stream plus the current DOM node. Default: 32.
MemoryBudgetGB int `mapstructure:"memory_budget_gb"`
// PerEntryEstimateMB is the conservative per-reader working-set estimate
// used to clamp auto ExtractorConcurrency against MemoryBudgetGB.
// Default: 256.
PerEntryEstimateMB int `mapstructure:"per_entry_estimate_mb"`
// BatchSize is the number of PatentRecords per sink Write call.
// Default: 1000.
BatchSize int `mapstructure:"batch_size"`
// BatchTimeout is the maximum time a partial batch waits before
// being flushed to the sink. Default: 2s.
BatchTimeout time.Duration `mapstructure:"batch_timeout"`
// OutputParquet is the destination Parquet file for `process`.
// Default: "./data.parquet".
OutputParquet string `mapstructure:"output_parquet"`
// RowGroupSize overrides DefaultRowGroupSize for the Parquet writer.
// 0 uses the library default.
RowGroupSize int `mapstructure:"row_group_size"`
// SpoolDir is where zip archives are spooled to disk (zip requires
// random access). Empty defaults to the OS temp directory.
SpoolDir string `mapstructure:"spool_dir"`
// UseLocalDir, when non-empty, replays archives from this local
// directory instead of fetching them over HTTP. Useful for re-runs
// after a prior run with KeepArchive=true.
UseLocalDir string `mapstructure:"use_local_dir"`
// KeepArchive tees each raw HTTP body to a file under ArchiveDir
// while the walker reads it. Orthogonal to KeepExtracted.
// Default: false.
KeepArchive bool `mapstructure:"keep_archive"`
// ArchiveDir is the destination directory when KeepArchive is true.
ArchiveDir string `mapstructure:"archive_dir"`
// KeepExtracted tees each selected archive entry to a file under
// ExtractedDir while the consumer reads it. Default: false.
KeepExtracted bool `mapstructure:"keep_extracted"`
// ExtractedDir is the destination directory when KeepExtracted is true.
ExtractedDir string `mapstructure:"extracted_dir"`
// CheckpointDB, when non-empty, enables resumable processing via a
// bbolt database at this path. Completed archives are recorded so
// subsequent runs skip them automatically.
CheckpointDB string `mapstructure:"checkpoint_db"`
// ResetCheckpoint, when true, deletes CheckpointDB and any existing
// OutputParquet shards at startup so the next run starts from scratch.
ResetCheckpoint bool `mapstructure:"reset_checkpoint"`
}
Pipeline configures the streaming pipeline shared by both subcommands.
type Server ¶
type Server struct {
// BaseURL is the root of the EPO BDDS REST API, e.g.
// "https://ops.epo.org/3.2/rest-services".
BaseURL string `mapstructure:"base_url" validate:"omitempty,url"`
// Timeout is the per-request HTTP timeout. Default: 30s.
Timeout time.Duration `mapstructure:"timeout" validate:"omitempty,gt=0"`
// MaxRetries is the maximum number of HTTP retries per archive.
// Range: 0–10. Default: 3.
MaxRetries int `mapstructure:"max_retries" validate:"min=0,max=10"`
// ProductID selects the EPO bulk-data product to process.
// Default: 3 (EP front-text exchange data).
ProductID int `mapstructure:"product_id"`
// VerifySHA1 enables SHA-1 checksum verification for every download.
// Mismatches trigger a retry. Default: false.
VerifySHA1 bool `mapstructure:"verify_sha1"`
}
Server points the EPO product source at the live BDDS API. Only `process` reads it; `process-hupd` is unaffected, hence optional.