spb/uqo-research-db

1# ============================================================2# SPID — configuration & paths3# ============================================================4from __future__ import annotations56import logging7import os8from dataclasses import dataclass, field9from functools import lru_cache10from pathlib import Path1112# project-db/  (this file lives in project-db/spid/config.py)13PROJECT_ROOT = Path(__file__).resolve().parent.parent141516def _load_dotenv() -> None:17    """Minimal .env loader (no python-dotenv dependency required)."""18    env_path = PROJECT_ROOT / ".env"19    if not env_path.exists():20        return21    for line in env_path.read_text().splitlines():22        line = line.strip()23        if not line or line.startswith("#") or "=" not in line:24            continue25        key, _, val = line.partition("=")26        key, val = key.strip(), val.strip().strip('"').strip("'")27        os.environ.setdefault(key, val)282930_load_dotenv()31# The parent UQO_RESEARCH_DB that holds the parsed corpus.32RESEARCH_ROOT = PROJECT_ROOT.parent33SOURCE_DATA = RESEARCH_ROOT / "data"343536def _parse_scope(spec: str) -> dict:37    out: dict = {}38    for part in (spec or "").split(";"):39        part = part.strip()40        if not part or ":" not in part:41            continue42        form, items = part.split(":", 1)43        out[form.strip()] = [i.strip() for i in items.split(",") if i.strip()]44    return out454647@dataclass(frozen=True)48class Settings:49    """Resolved runtime configuration for the SPID platform."""5051    # --- source corpus (read-only) -------------------------------------52    source_data: Path = SOURCE_DATA53    parquet_dir: Path = SOURCE_DATA / "parquet"54    processed_dir: Path = SOURCE_DATA / "processed"5556    @property57    def filings_parquet(self) -> Path:58        return self.parquet_dir / "filings" / "filings.parquet"5960    @property61    def sections_parquet(self) -> Path:62        return self.parquet_dir / "filing_sections" / "filing_sections.parquet"6364    @property65    def companies_parquet(self) -> Path:66        return self.parquet_dir / "companies" / "companies.parquet"6768    # --- SPID outputs (writable) ---------------------------------------69    db_path: Path = PROJECT_ROOT / "data" / "db" / "spid.duckdb"70    out_dir: Path = PROJECT_ROOT / "data"7172    # --- extraction ----------------------------------------------------73    backend: str = field(74        default_factory=lambda: os.environ.get("SPID_BACKEND", "llm")75    )  # "llm" (gpt-4o-mini) | "rules"76    min_confidence: float = 0.4577    # which parsed sections feed the extractor, per form (None/empty = every section)78    section_scope: dict = field(default_factory=lambda: _parse_scope(os.environ.get(79        "SPID_SECTION_SCOPE",80        "10-K:business,risk_factors,properties,md_and_a,quantitative_disclosures")))81    max_section_chars: int = 600_0008283    # --- embeddings ----------------------------------------------------84    embedding_model: str = field(85        default_factory=lambda: os.environ.get("SPID_EMBED_MODEL", "all-MiniLM-L6-v2")86    )8788    # --- LLM backend (OpenAI-compatible) -------------------------------89    llm_base_url: str = field(90        default_factory=lambda: os.environ.get(91            "SPID_LLM_BASE_URL", "https://api.openai.com/v1"92        )93    )94    llm_model: str = field(95        default_factory=lambda: os.environ.get("SPID_LLM_MODEL", "gpt-4o-mini")96    )97    llm_api_key: str = field(98        default_factory=lambda: os.environ.get("OPENAI_API_KEY", "")99        or os.environ.get("MACLUSTR_API_KEY", "")100    )101    llm_timeout: float = 120.0102103    def ensure_dirs(self) -> None:104        self.db_path.parent.mkdir(parents=True, exist_ok=True)105        (self.out_dir / "exports").mkdir(parents=True, exist_ok=True)106107108@lru_cache(maxsize=1)109def get_settings() -> Settings:110    s = Settings()111    s.ensure_dirs()112    return s113114115def get_logger(name: str) -> logging.Logger:116    logger = logging.getLogger(f"spid.{name}")117    if not logger.handlers:118        h = logging.StreamHandler()119        h.setFormatter(120            logging.Formatter("%(asctime)s %(levelname)-7s %(name)s | %(message)s",121                              datefmt="%H:%M:%S")122        )123        logger.addHandler(h)124        logger.setLevel(os.environ.get("SPID_LOG_LEVEL", "INFO"))125        logger.propagate = False126    return logger127