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 

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

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

19 

20_TRACE_ENV = "LILBEE_INGEST_TRACE" 

21 

22 

23@dataclass(frozen=True) 

24class ExtractionTrace: 

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

26 

27 source: str 

28 content_type: str 

29 elapsed_s: float 

30 page_count: int 

31 chunk_count: int 

32 ocr_pages: int 

33 vision_configured: bool 

34 

35 @property 

36 def used_vision(self) -> bool: 

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

38 return self.ocr_pages > 0 and self.vision_configured 

39 

40 def as_line(self) -> str: 

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

42 return ( 

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

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

45 f"chunks={self.chunk_count} ocr_pages={self.ocr_pages} " 

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

47 ) 

48 

49 

50def configure_from_env() -> None: 

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

52 LILBEE_INGEST_TRACE_FILE is set. 

53 

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

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

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

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

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

59 return 

60 trace_log.setLevel(logging.DEBUG) 

61 vision_log.setLevel(logging.INFO) 

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

63 if not target: 

64 return 

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

66 for logger in (trace_log, vision_log): 

67 if any( 

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

69 for h in logger.handlers 

70 ): 

71 continue 

72 handler = logging.FileHandler(resolved) 

73 handler.setLevel(logging.DEBUG) 

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

75 logger.addHandler(handler) 

76 

77 

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

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

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

81 if trace.used_vision: 

82 vision_log.info( 

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

84 trace.source, 

85 trace.ocr_pages, 

86 trace.elapsed_s * 1000, 

87 )