# ============================================================ # Project: EDGAR Research Database # Description: SEC EDGAR Research Infrastructure # # Authors: # - Manel Kammoun — manel.kammoun@uqo.ca # - Djerrycharli Tandjambianda — djerrycharli.tandjambianda@uqo.ca # - Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca # # Institution: # Université du Québec en Outaouais (UQO) # # License: # Research / Academic Use # ============================================================ from __future__ import annotations import os from functools import lru_cache from pathlib import Path import yaml from dotenv import load_dotenv load_dotenv() def get_project_root() -> Path: return Path(__file__).resolve().parent.parent.parent @lru_cache(maxsize=None) def _load_yaml(path: Path) -> dict: with open(path) as f: raw = yaml.safe_load(f) or {} return _resolve_env_vars(raw) def _resolve_env_vars(obj): if isinstance(obj, str) and obj.startswith("${") and obj.endswith("}"): var_name = obj[2:-1] return os.environ.get(var_name, obj) if isinstance(obj, dict): return {k: _resolve_env_vars(v) for k, v in obj.items()} if isinstance(obj, list): return [_resolve_env_vars(item) for item in obj] return obj def get_config(name: str | None = None) -> dict: root = get_project_root() config_dir = root / "config" if name: path = config_dir / f"{name}.yaml" if not path.exists(): raise FileNotFoundError(f"Config file not found: {path}") return _load_yaml(path) merged = {} for yaml_file in sorted(config_dir.glob("*.yaml")): merged[yaml_file.stem] = _load_yaml(yaml_file) return merged def get_sec_config() -> dict: return get_config("sec").get("sec", {}) def get_database_config() -> dict: return get_config("database").get("database", {}) def get_parsing_config() -> dict: return get_config("parsing").get("parsing", {}) def get_embeddings_config() -> dict: return get_config("embeddings").get("embeddings", {}) def ensure_data_dirs(): from src.utils.constants import DATA_DIRS for d in DATA_DIRS.values(): d.mkdir(parents=True, exist_ok=True)