1#!/usr/bin/env python3
2""" """
3
4# Imports:
5from __future__ import annotations
6
7# ##-- stdlib imports
8import builtins
9import datetime
10import functools as ftz
11import itertools as itz
12import logging as logmod
13import os
14import re
15import time
16import types
17import warnings
18import weakref
19from sys import stderr, stdout
20from uuid import UUID, uuid1
21
22# ##-- end stdlib imports
23
24# ##-- 1st party imports
25from jgdv import Mixin, Proto
26from jgdv.structs.metalord.singleton import MLSingleton
27from jgdv.structs.chainguard import ChainGuard
28# ##-- end 1st party imports
29
30from . import _interface as API # noqa: N812
31from .format import StripColourFormatter
32from .logger import JGDVLogger
33from .logger_spec import LoggerSpec
34
35# ##-- types
36# isort: off
37import abc
38import collections.abc
39from typing import TYPE_CHECKING, cast, assert_type, assert_never, override
40from typing import Generic, NewType
41
42# Protocols:
43from typing import Protocol, runtime_checkable
44
45# Typing Decorators:
46from typing import no_type_check, final, overload
47
48if TYPE_CHECKING:
49 import pathlib as pl
50 import enum
51 from jgdv import Maybe
52 from typing import Final
53 from typing import ClassVar, Any, LiteralString
54 from typing import Never, Self, Literal
55 from typing import TypeGuard
56 from collections.abc import Iterable, Iterator, Callable, Generator
57 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
58 from ._interface import Logger
59
60##--|
61
62# isort: on
63# ##-- end types
64
65##-- logging
66logging = logmod.getLogger(__name__)
67##-- end logging
68
69PRINTER_INITIAL_SPEC : Final[LoggerSpec] = LoggerSpec.build(API.default_printer)
70INITIAL_SPEC : Final[LoggerSpec] = LoggerSpec.build(API.default_stdout)
71
72##--|
73
[docs]
74class PrintCapture_m:
75 """Mixin for redirecting builtins.print to a file"""
76
77 original_print: Maybe[Callable]
78
[docs]
79 def capture_printing_to_file(self, path: str | pl.Path = API.default_print_file, *, disable_warning: bool=False) -> None:
80 """Modifies builtins.print to also print to a file
81
82 Setup a file handler for a separate logger,
83 to keep a trace of anything printed.
84 Strips colour print command codes out of any string
85 printed strings are logged at DEBUG level
86 """
87 match getattr(self, "original_print", None):
88 case None:
89 self.original_print = builtins.print
90 case _:
91 return
92
93 if not disable_warning:
94 warnings.warn("Modifying builtins.print", RuntimeWarning, 2)
95
96 file_handler = logmod.FileHandler(path, mode="w")
97 file_handler.setLevel(logmod.DEBUG)
98 file_handler.setFormatter(StripColourFormatter())
99
100 print_logger = logmod.getLogger(f"{API.PRINTER_NAME}.intercept")
101 print_logger.setLevel(logmod.DEBUG)
102 print_logger.addHandler(file_handler)
103 print_logger.propagate = False
104
105 @ftz.wraps(self.original_print)
106 def intercepted(*args: Any, **kwargs: Any) -> None: # noqa: ANN401
107 """Wraps `print` to also log to a separate file"""
108 assert self.original_print is not None
109 self.original_print(*args, **kwargs)
110 if bool(args):
111 print_logger.debug(args[0])
112
113 builtins.print = intercepted
114
[docs]
115 def remove_print_capture(self) -> None:
116 """removes a previously advised builtins.print"""
117 match getattr(self, "original_print", None):
118 case None:
119 return
120 case x:
121 builtins.print = x
122 self.original_print = None
123
124##--|
125
[docs]
126@Mixin(PrintCapture_m)
127class JGDVLogConfig(metaclass=MLSingleton):
128 """Utility class to setup [stdout, stderr, file] logging.
129
130 Also creates a 'printer' logger, so instead of using `print`,
131 tasks can notify the user using the printer,
132 which also includes the notifications into the general log trace
133
134 The Printer has a number of children, which can be controlled
135 to customise verbosity.
136
137 Standard _printer children::
138 action_exec, action_group, artifact, cmd, fail, header, help, queue,
139 report, skip, sleep, success, task, task_header, task_loop, task_state,
140 track
141
142 """
143
144 ##--| classvars
145 levels : ClassVar[type[API.LogLevel_e]] = API.LogLevel_e
146 logger_cls : ClassVar[type[Logger]] = JGDVLogger
147 record_cls : ClassVar[type[logmod.LogRecord]] = logmod.LogRecord
148 ##--| core vars
149 root : Logger
150 is_setup : bool
151 ##--| internal vars
152 _initial_spec : LoggerSpec
153 _printer_initial_spec : LoggerSpec
154 _registry : dict[str, LoggerSpec]
155
156 def __init__(self) -> None:
157 # Root Logger for everything
158 self.root = logmod.root
159 self.is_setup = False
160 self._initial_spec = INITIAL_SPEC
161 self._printer_initial_spec = PRINTER_INITIAL_SPEC
162 self._registry = {}
163
164 self._install_logger_override()
165 self._register_new_names()
166
167 self.activate_spec(self._initial_spec)
168 self.activate_spec(self._printer_initial_spec)
169
170 logging.log(self.levels.bootstrap, "Post Log Setup")
171
172 ##--| dunders
173
174 @override
175 def __repr__(self) -> str:
176 return f"<{self.__class__.__name__}({len(self._registry)}) : >"
177
178 ##--| internal
179
[docs]
180 def _install_logger_override(self) -> None:
181 match self.logger_cls:
182 case None:
183 return
184 case x if not hasattr(x, "install"):
185 return
186 case x:
187 x.install()
188 if not issubclass(logmod.getLoggerClass(), self.logger_cls):
189 msg = "Logger Class Installation Failed"
190 raise TypeError(msg, self.logger_cls)
191
[docs]
192 def _install_record_override(self) -> None:
193 """ Sets a custom LogRecord subclass as the factory """
194 logmod.setLogRecordFactory(self.record_cls)
195
[docs]
196 def _register_new_names(self) -> None:
197 for name, lvl in self.levels.__members__.items():
198 logmod.addLevelName(lvl, name)
199
211
212 ##--| public
213
[docs]
214 def report(self) -> None:
215 """Utility method to inspect the state of logging."""
216 printer = self.subprinter()
217 for x in self._registry.values():
218 printer.log(logmod.WARN, "%s : %s", x.fullname, x.level)
219
[docs]
220 def activate_spec(self, spec: LoggerSpec, *, override: bool = False) -> None: # noqa: ARG002
221 """Add a spec to the registry and activate it"""
222 target = spec
223 fullname = spec.fullname
224 logging.info("Activating Logging Spec: %s", fullname)
225 self._registry[fullname] = spec
226 target.apply()
227
[docs]
228 def setup(self, config: dict | ChainGuard, *, force: bool = False) -> None:
229 """a setup that uses config values"""
230 if self.is_setup and not force:
231 warnings.warn("Logging Is Already Set Up", stacklevel=2)
232
233 match config:
234 case x if not bool(x):
235 msg = "Config data has not been configured"
236 raise ValueError(msg)
237 case dict():
238 config = ChainGuard(config)
239 case ChainGuard():
240 pass
241
242 assert isinstance(config, ChainGuard)
243 self._initial_spec.clear()
244 self._printer_initial_spec.clear()
245
246 root_spec = LoggerSpec.build(
247 [
248 config.on_fail({}).logging.file(), # type: ignore[attr-defined]
249 config.on_fail({}).logging.stream(), # type: ignore[attr-defined]
250 ],
251 name=LoggerSpec.RootName,
252 )
253 print_spec = LoggerSpec.build(
254 config.on_fail({}).logging.printer(), # type: ignore[attr-defined]
255 name=API.PRINTER_NAME,
256 )
257
258 self.activate_spec(root_spec, override=True)
259 self.activate_spec(print_spec, override=True)
260 self._setup_logging_extra(config)
261
262 self.is_setup = True
263
[docs]
264 def reset(self) -> None:
265 """
266 Reset the config to the initial specs
267 """
268 self.activate_spec(self._initial_spec)
269 self.activate_spec(self._printer_initial_spec)
270
[docs]
271 def set_level(self, level: int | str) -> None:
272 """Set the active logging level"""
273 names = logmod.getLevelNamesMapping()
274 lvl = None
275 match level:
276 case int():
277 lvl = level
278 case str() if (lvl := names.get(level, None)) is not None:
279 pass
280 case _:
281 msg = "Unknown Level Name"
282 raise ValueError(msg, level)
283
284 assert lvl is not None
285 self._initial_spec.set_level(lvl)
286 self._printer_initial_spec.set_level(lvl)
287
[docs]
288 def subprinter(self, *names: str, prefix: Maybe[str] = None) -> Logger:
289 """Get a subprinter of the printer logger.
290 The First name needs to be a registered subprinter.
291 Additional names are unconstrained
292 """
293 base = self._printer_initial_spec.get()
294 if not bool(names) or names == (None,):
295 return base
296
297 current = base
298 for name in names:
299 current = current.getChild(name)
300 else:
301 if prefix and hasattr(current, "set_prefixes"):
302 current.set_prefixes(prefix)
303 return current