spb/uqo-research-db

1# ============================================================2# Project: EDGAR Research Database3# Description: SEC EDGAR Research Infrastructure4#5# Authors:6# - Manel Kammoun — manel.kammoun@uqo.ca7# - Djerrycharli Tandjambianda — djerrycharli.tandjambianda@uqo.ca8# - Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca9#10# Institution:11# Université du Québec en Outaouais (UQO)12#13# License:14# Research / Academic Use15# ============================================================1617from __future__ import annotations1819import json20import logging21import os22import sys23from datetime import datetime, timezone24from pathlib import Path252627class JsonFormatter(logging.Formatter):28    def format(self, record: logging.LogRecord) -> str:29        log_entry = {30            "timestamp": datetime.now(timezone.utc).isoformat(),31            "level": record.levelname,32            "logger": record.name,33            "message": record.getMessage(),34        }35        if record.exc_info and record.exc_info[1]:36            log_entry["exception"] = self.formatException(record.exc_info)37        for key in ("filing", "cik", "accession", "form_type", "step"):38            val = getattr(record, key, None)39            if val is not None:40                log_entry[key] = val41        return json.dumps(log_entry)424344def setup_logging(45    name: str = "edgar",46    level: str | None = None,47    log_dir: Path | None = None,48    log_format: str | None = None,49) -> logging.Logger:50    level = level or os.environ.get("LOG_LEVEL", "INFO")51    log_format = log_format or os.environ.get("LOG_FORMAT", "text")5253    logger = logging.getLogger(name)54    logger.setLevel(getattr(logging, level.upper(), logging.INFO))5556    if logger.handlers:57        return logger5859    console = logging.StreamHandler(sys.stdout)60    if log_format == "json":61        console.setFormatter(JsonFormatter())62    else:63        console.setFormatter(64            logging.Formatter("%(asctime)s [%(levelname)-8s] %(name)s — %(message)s")65        )66    logger.addHandler(console)6768    if log_dir:69        log_dir.mkdir(parents=True, exist_ok=True)70        today = datetime.now().strftime("%Y-%m-%d")71        file_handler = logging.FileHandler(log_dir / f"{name}_{today}.log")72        file_handler.setFormatter(JsonFormatter())73        logger.addHandler(file_handler)7475    return logger767778def get_logger(name: str) -> logging.Logger:79    return logging.getLogger(f"edgar.{name}")80