02.11.04 — Loggers: PrintLogger, FileLogger, PrintAndFileLogger, MutedLogger#

LOGGERS_CHOICES = ("terminal", "file", "terminal+file", "muted")

def create_matching_logger(mode: SupportedLogger) -> ILogger:
    if mode == "terminal":
        return PrintLogger()
    if mode == "file":
        return FileLogger(base_dir=get_default_log_dir())
    if mode == "terminal+file":
        return PrintAndFileLogger(base_dir=get_default_log_dir())
    if mode == "muted":
        return MutedLogger()
    raise ValueError(f"Unsupported logger mode: {mode}")

def get_default_log_dir() -> Path:
    return Path.cwd() / ".ocarina_logs"

class PrintLogger(ILogger):
    def __init__(self) -> None:
        self._prefix_thunk: Thunk[str] = lambda: ""
        self._domain_taxonomy: tuple[str, ...] = ("",)

    def set_prefix(self, prefix_thunk: Thunk[str]) -> Self:
        self._prefix_thunk = prefix_thunk
        return self

    def set_domain_taxonomy(self, taxonomy: tuple[str, ...]) -> Self:
        self._domain_taxonomy = taxonomy
        return self

    def raw(self, *args, stream=None, **kwargs) -> None:
        if stream is None:
            stream = sys.stdout
        print(*args, file=stream, **kwargs)

    def info(self, msg, *args, exc=None, **kwargs):
        self._print_log("[INFO]", msg, *args, exc=exc, stream=sys.stdout)
    # ... id. for critical / error / warning / debug / success / test_name ...

@final
class FileLogger(PrintLogger):
    def __init__(
        self, *,
        base_dir: Path,
        with_flush_effect: bool = True,
        with_fallback_on_print_logger_when_no_taxonomy_effect: bool = True,
    ) -> None:
        self._base_dir = base_dir
        self._with_flush_effect = with_flush_effect
        self._with_fallback_on_print_logger_when_no_taxonomy_effect = (
            with_fallback_on_print_logger_when_no_taxonomy_effect
        )
        super().__init__()

    @property
    def _log_file_path(self) -> Path:
        if len(self._domain_taxonomy) == 1:
            folder_path = self._base_dir
            file_name = f"{self._domain_taxonomy[0]}.log"
        else:
            folder_path = self._base_dir.joinpath(*self._domain_taxonomy[:-1])
            file_name = f"{self._domain_taxonomy[-1]}.log"
        folder_path.mkdir(parents=True, exist_ok=True)
        return folder_path / file_name

class PrintAndFileLogger(FileLogger):
    # … inherits from FileLogger but ALSO prints to stdout/stderr …

class MutedLogger(ILogger):
    def set_prefix(self, prefix_thunk): return self
    def set_domain_taxonomy(self, taxonomy): return self
    def raw(self, *args, stream=None, **kwargs): pass
    def info(self, msg, *args, exc=None, **kwargs): pass
    # … etc, everything is pass …
    def cleanup(self): pass

# src/ocarina/opinionated/loggers/utils/format_metadata_str.py
def _format_metadata_str(*, namespace: str, metadata: str) -> str:
    return "[" + namespace + "::" + metadata + "]"


def format_utc_date_metadata_str() -> str:
    return _format_metadata_str(
        namespace="UTC_DATE",
        metadata=datetime.now(UTC).isoformat(),
    )


def format_current_thread_metadata_str() -> str:
    return _format_metadata_str(namespace="THREAD", metadata=current_thread().name)


def concat_metadata(*formatters: Thunk[str]) -> str:
    ...

_DEFAULT_UTC_DATE_REGEX = re.compile(r"\[UTC_DATE::([^]]+)]")

def _replace_utc_date(line: str, *, utc_date_regex: re.Pattern[str]) -> str:
    def _repl(m: re.Match[str]) -> str:
        return (
            datetime.fromisoformat(m.group(1).replace("Z", "+00:00"))
            .astimezone()
            .strftime("[%m/%d/%Y | %Hh%M:%S.%f]")
        )
    return utc_date_regex.sub(_repl, line)

exceptions_logger = PrintLogger()
    .set_prefix(
        lambda: concat_metadata(
            format_utc_date_metadata_str,
            format_current_thread_metadata_str,
        )
    )
    .set_domain_taxonomy(("Post-execution plugins",))

def cleanup(self) -> None:
    # Flush effect (optionally print to stdout)
    # Close file handle
    # Recycle for next attempt
    ...