Coverage for documents/management/commands/document_sanity_checker.py: 0%

1"""Management command to check the document archive for issues.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6from typing import Any 

7 

8from rich.panel import Panel 

9from rich.table import Table 

10from rich.text import Text 

11 

12from documents.management.commands.base import PaperlessCommand 

13from documents.models import Document 

14from documents.sanity_checker import SanityCheckMessages 

15from documents.sanity_checker import check_sanity 

16 

17_LEVEL_STYLE: dict[int, tuple[str, str]] = { 

18 logging.ERROR: ("bold red", "ERROR"), 

19 logging.WARNING: ("yellow", "WARN"), 

20 logging.INFO: ("dim", "INFO"), 

21} 

22 

23 

24class Command(PaperlessCommand): 

25 help = "This command checks your document archive for issues." 

26 

27 supports_progress_bar = True 

28 supports_multiprocessing = False 

29 

30 def _render_results(self, messages: SanityCheckMessages) -> None: 

31 """Render sanity check results as a Rich table.""" 

32 

33 if ( 

34 not messages.has_error 

35 and not messages.has_warning 

36 and not messages.has_info 

37 ): 

38 self.console.print( 

39 Panel( 

40 "[green]No issues detected.[/green]", 

41 title="Sanity Check", 

42 border_style="green", 

43 ), 

44 ) 

45 return 

46 

47 # Build a lookup for document titles 

48 doc_pks = [pk for pk in messages.document_pks() if pk is not None] 

49 titles: dict[int, str] = {} 

50 if doc_pks: 

51 titles = dict( 

52 Document.global_objects.filter(pk__in=doc_pks) 

53 .only("pk", "title") 

54 .values_list("pk", "title"), 

55 ) 

56 

57 table = Table( 

58 title="Sanity Check Results", 

59 show_lines=True, 

60 title_style="bold", 

61 ) 

62 table.add_column("Level", width=7, no_wrap=True) 

63 table.add_column("Document", min_width=20) 

64 table.add_column("Issue", ratio=1, overflow="fold") 

65 

66 for doc_pk, doc_messages in messages.iter_messages(): 

67 if doc_pk is not None: 

68 title = titles.get(doc_pk, "Unknown") 

69 doc_label = f"#{doc_pk} {title}" 

70 else: 

71 doc_label = "(global)" 

72 

73 for msg in doc_messages: 

74 style, label = _LEVEL_STYLE.get( 

75 msg["level"], 

76 ("dim", "INFO"), 

77 ) 

78 table.add_row( 

79 Text(label, style=style), 

80 Text(doc_label), 

81 Text(str(msg["message"])), 

82 ) 

83 

84 self.console.print(table) 

85 

86 parts: list[str] = [] 

87 

88 if messages.document_error_count: 

89 parts.append( 

90 f"{messages.document_error_count} document(s) with [bold red]errors[/bold red]", 

91 ) 

92 if messages.document_warning_count: 

93 parts.append( 

94 f"{messages.document_warning_count} document(s) with [yellow]warnings[/yellow]", 

95 ) 

96 if messages.document_info_count: 

97 parts.append(f"{messages.document_info_count} document(s) with infos") 

98 if messages.global_warning_count: 

99 parts.append( 

100 f"{messages.global_warning_count} global [yellow]warning(s)[/yellow]", 

101 ) 

102 

103 if parts: 

104 if len(parts) > 1: 

105 summary = ", ".join(parts[:-1]) + " and " + parts[-1] 

106 else: 

107 summary = parts[0] 

108 self.console.print(f"\nFound {summary}.") 

109 else: 

110 self.console.print("\nNo issues found.") 

111 

112 def handle(self, *args: Any, **options: Any) -> None: 

113 messages = check_sanity( 

114 iter_wrapper=lambda docs: self.track( 

115 docs, 

116 description="Checking documents...", 

117 ), 

118 ) 

119 self._render_results(messages)