Coverage for src/lilbee/data/extract/trace.py: 100%

1"""Structured per-file extraction tracing, for sharing ingest diagnostics. 

2 

3Every xberg extraction emits one machine-parseable line on the ``lilbee.ingest.trace`` 

4logger: filename, wall-clock, chunk and page counts, and how many pages fell through 

5to OCR. Files that needed the vision model also emit a line on ``lilbee.ingest.vision`` 

6so ``grep vision`` yields exactly the set of scanned files. Enable with 

7``LILBEE_INGEST_TRACE=1`` (sets both loggers to DEBUG). 

8""" 

9 

10from __future__ import annotations 

11 

12import logging 

13import os 

14from dataclasses import dataclass 

15from pathlib import Path 

16 

17from lilbee.data.types import OcrReport 

18from lilbee.runtime.progress import OcrBackendUsed 

19 

20trace_log = logging.getLogger("lilbee.ingest.trace") 

21vision_log = logging.getLogger("lilbee.ingest.vision") 

22 

23_TRACE_ENV = "LILBEE_INGEST_TRACE" 

24 

25 

26@dataclass(frozen=True) 

27class ExtractionTrace: 

28 """One xberg extraction's measured outcome.""" 

29 

30 source: str 

31 content_type: str 

32 elapsed_s: float 

33 page_count: int 

34 chunk_count: int 

35 ocr: OcrReport 

36 

37 @property 

38 def used_vision(self) -> bool: 

39 """A page fell through to OCR and the OCR backend is the vision model.""" 

40 return self.ocr.pages > 0 and self.ocr.backend is OcrBackendUsed.VISION 

41 

42 def as_line(self) -> str: 

43 """A stable key=value line, easy to grep, diff, and hand to the xberg author.""" 

44 return ( 

45 f"extract source={self.source!r} type={self.content_type} " 

46 f"elapsed_ms={self.elapsed_s * 1000:.0f} pages={self.page_count} " 

47 f"chunks={self.chunk_count} ocr={self.ocr.backend} ocr_pages={self.ocr.pages} " 

48 f"vision={'yes' if self.used_vision else 'no'}" 

49 ) 

50 

51 

52def configure_from_env() -> None: 

53 """Enable trace/vision logging per LILBEE_INGEST_TRACE; mirror to a file when 

54 LILBEE_INGEST_TRACE_FILE is set. 

55 

56 The file handler exists because host apps own the root handlers: the TUI 

57 logs WARNING+ to its file, so INFO trace lines vanish there even with the 

58 loggers enabled. A dedicated handler on these two loggers makes the trace 

59 destination independent of whichever front-end is running.""" 

60 if os.environ.get("LILBEE_INGEST_TRACE", "").strip().lower() not in {"1", "true", "yes"}: 

61 return 

62 trace_log.setLevel(logging.DEBUG) 

63 vision_log.setLevel(logging.INFO) 

64 target = os.environ.get("LILBEE_INGEST_TRACE_FILE", "").strip() 

65 if not target: 

66 return 

67 resolved = str(Path(target).absolute()) 

68 for logger in (trace_log, vision_log): 

69 if any( 

70 isinstance(h, logging.FileHandler) and h.baseFilename == resolved 

71 for h in logger.handlers 

72 ): 

73 continue 

74 handler = logging.FileHandler(resolved) 

75 handler.setLevel(logging.DEBUG) 

76 handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")) 

77 logger.addHandler(handler) 

78 

79 

80def trace_extraction(trace: ExtractionTrace) -> None: 

81 """Log one extraction's outcome, plus a dedicated line if it needed vision.""" 

82 trace_log.info("%s", trace.as_line()) 

83 if trace.used_vision: 

84 vision_log.info( 

85 "vision-ocr source=%r ocr_pages=%d elapsed_ms=%.0f", 

86 trace.source, 

87 trace.ocr.pages, 

88 trace.elapsed_s * 1000, 

89 )