feat: json log

beet/contrib/json_log.py

@@ -0,0 +1,74 @@
+"""Plugin for collecting json logs."""
+
+
+__all__ = [
+    "JsonLogHandler",
+    "JsonLogEntry",
+    "activate_json_log",
+]
+
+
+import logging
+from contextlib import contextmanager
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+from beet import Context
+from beet.contrib.json_reporter import JsonReporter
+from beet.toolchain.cli import LogHandler
+
+
+class JsonLogEntry(BaseModel):
+    level: str
+    prefix: str
+    message: str
+    annotation: Optional[str]
+    details: List[str]
+
+
+def activate_json_log(ctx: Context):
+    with ctx.inject(JsonLogHandler).activate():
+        yield
+
+
+def beet_default(ctx: Context):
+    json_reporter = ctx.inject(JsonReporter)
+    json_reporter.data["log"] = [
+        entry.dict() for entry in ctx.inject(JsonLogHandler).entries
+    ]
+
+
+class JsonLogHandler(LogHandler):
+    """Logging handler that collects log records as json."""
+
+    entries: List[JsonLogEntry]
+
+    def __init__(self, ctx: Optional[Context] = None):
+        super().__init__()
+        self.entries = []
+
+    @contextmanager
+    def activate(self):
+        logger = logging.getLogger()
+        previous_handlers = logger.handlers
+        logger.handlers = [self]
+
+        try:
+            yield
+        finally:
+            logger.handlers = previous_handlers
+
+    def emit(self, record: logging.LogRecord):
+        level = self.abbreviations.get(record.levelname, record.levelname)
+        leading_line, *lines = self.format(record).splitlines()
+
+        self.entries.append(
+            JsonLogEntry(
+                level=level,
+                prefix=getattr(record, "prefix", record.name),
+                message=leading_line,
+                annotation=getattr(record, "annotate", None),
+                details=lines,
+            )
+        )

beet/contrib/json_reporter.py

@@ -19,7 +19,7 @@
 from contextlib import contextmanager, redirect_stdout
 from dataclasses import dataclass, field
 from functools import cached_property
-from typing import Any, List, Optional, Union
+from typing import Any, Iterator, List, Optional, Union
 
 from pydantic import BaseModel
 
@@ -43,6 +43,7 @@ class JsonReporterOptions(BaseModel):
     exception_filter: Optional[ListOption[str]] = None
     handlers: List[str] = [
         "beet.contrib.json_reporter.stdout",
+        "beet.contrib.json_log",
         "beet.contrib.json_reporter.resource_pack_listing",
         "beet.contrib.json_reporter.resource_pack_zip",
         "beet.contrib.json_reporter.data_pack_listing",
@@ -51,7 +52,9 @@ class JsonReporterOptions(BaseModel):
 
 
 def beet_default(ctx: Context):
-    with ctx.inject(JsonReporter).activate():
+    with ctx.inject(JsonReporter).activate() as json_reporter:
+        if json_reporter.opts.enabled:
+            ctx.require("beet.contrib.json_log.activate_json_log")
         yield
 
 
@@ -75,17 +78,17 @@ def add_handler(self, *specs: PluginSpec):
         self.handlers.extend(specs)
 
     @contextmanager
-    def activate(self):
+    def activate(self) -> Iterator["JsonReporter"]:
         if not self.opts.enabled:
-            yield
+            yield self
             return
 
         message = None
         exception = None
 
         try:
             with redirect_stdout(self.stdout):
-                yield
+                yield self
         except WrappedException as exc:
             message = str(exc)
             if not exc.hide_wrapped_exception: