Coverage for documents/sanity_checker.py: 18%
156 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 09:07 +0000
1"""
2Sanity checker for the Paperless-ngx document archive.
4Verifies that all documents have valid files, correct checksums,
5and consistent metadata. Reports orphaned files in the media directory.
7Progress display is the caller's responsibility -- pass an ``iter_wrapper``
8to wrap the document queryset (e.g., with a progress bar). The default
9is an identity function that adds no overhead.
10"""
12import logging
13from collections import defaultdict
14from collections.abc import Iterator
15from pathlib import Path
16from typing import TYPE_CHECKING
17from typing import Final
18from typing import TypedDict
20from django.conf import settings
22from documents.models import Document
23from documents.utils import IterWrapper
24from documents.utils import compute_checksum
25from documents.utils import identity
26from paperless.config import GeneralConfig
28logger = logging.getLogger("paperless.sanity_checker")
31class MessageEntry(TypedDict):
32 """A single sanity check message with its severity level."""
34 level: int
35 message: str
38class SanityCheckMessages:
39 """Collects sanity check messages grouped by document primary key.
41 Messages are categorized as error, warning, or info. ``None`` is used
42 as the key for messages not associated with a specific document
43 (e.g., orphaned files).
44 """
46 def __init__(self) -> None:
47 self._messages: dict[int | None, list[MessageEntry]] = defaultdict(list)
48 self._document_pks: set[int] = set()
49 self._document_error_pks: set[int] = set()
50 self._document_warning_pks: set[int] = set()
51 self._document_info_pks: set[int] = set()
52 self._document_error_issue_count: int = 0
53 self._document_warning_issue_count: int = 0
54 self.has_error: bool = False
55 self.has_warning: bool = False
56 self.has_info: bool = False
57 self.document_count: int = 0
58 self.document_error_count: int = 0
59 self.document_warning_count: int = 0
60 self.document_info_count: int = 0
61 self.global_warning_count: int = 0
63 # -- Recording ----------------------------------------------------------
65 def _add_document_issue(self, doc_pk: int, document_pks: set[int]) -> bool:
66 if doc_pk not in self._document_pks:
67 self._document_pks.add(doc_pk)
68 self.document_count += 1
70 if doc_pk in document_pks:
71 return False
73 document_pks.add(doc_pk)
74 return True
76 def error(self, doc_pk: int | None, message: str) -> None:
77 self._messages[doc_pk].append({"level": logging.ERROR, "message": message})
78 self.has_error = True
79 if doc_pk is not None:
80 self._document_error_issue_count += 1
81 if self._add_document_issue(doc_pk, self._document_error_pks):
82 self.document_error_count += 1
84 def warning(self, doc_pk: int | None, message: str) -> None:
85 self._messages[doc_pk].append({"level": logging.WARNING, "message": message})
86 self.has_warning = True
88 if doc_pk is not None:
89 self._document_warning_issue_count += 1
90 if self._add_document_issue(doc_pk, self._document_warning_pks):
91 self.document_warning_count += 1
92 else:
93 # This is the only type of global message we do right now
94 self.global_warning_count += 1
96 def info(self, doc_pk: int | None, message: str) -> None:
97 self._messages[doc_pk].append({"level": logging.INFO, "message": message})
98 self.has_info = True
100 if doc_pk is not None and self._add_document_issue(
101 doc_pk,
102 self._document_info_pks,
103 ):
104 self.document_info_count += 1
106 # -- Iteration / query --------------------------------------------------
108 def document_pks(self) -> list[int | None]:
109 """Return all document PKs (including None for global messages)."""
110 return list(self._messages.keys())
112 def iter_messages(self) -> Iterator[tuple[int | None, list[MessageEntry]]]:
113 """Iterate over (doc_pk, messages) pairs."""
114 yield from self._messages.items()
116 def __getitem__(self, item: int | None) -> list[MessageEntry]:
117 return self._messages[item]
119 # -- Summarize Helpers --------------------------------------------------
121 @property
122 def has_global_issues(self) -> bool:
123 return None in self._messages
125 @property
126 def total_issue_count(self) -> int:
127 """Total number of error and warning messages across all documents and global."""
128 return (
129 self._document_error_issue_count
130 + self._document_warning_issue_count
131 + self.global_warning_count
132 )
134 # -- Logging output (used by Celery task path) --------------------------
136 def log_messages(self) -> None:
137 """Write all messages to the ``paperless.sanity_checker`` logger.
139 This is the output path for headless / Celery execution.
140 Management commands use Rich rendering instead.
141 """
142 if len(self._messages) == 0:
143 logger.info("Sanity checker detected no issues.")
144 return
146 doc_pks = [pk for pk in self._messages if pk is not None]
147 titles: dict[int, str] = {}
148 if doc_pks:
149 titles = dict(
150 Document.global_objects.filter(pk__in=doc_pks)
151 .only("pk", "title")
152 .values_list("pk", "title"),
153 )
155 for doc_pk, entries in self._messages.items():
156 if doc_pk is not None:
157 title = titles.get(doc_pk, "Unknown")
158 logger.info(
159 "Detected following issue(s) with document #%s, titled %s",
160 doc_pk,
161 title,
162 )
163 for msg in entries:
164 logger.log(msg["level"], msg["message"])
167class SanityCheckFailedException(Exception):
168 pass
171# ---------------------------------------------------------------------------
172# Internal helpers
173# ---------------------------------------------------------------------------
176def _build_present_files() -> set[Path]:
177 """Collect all files in MEDIA_ROOT, excluding directories and ignorable files."""
178 present_files = {
179 x.resolve()
180 for x in Path(settings.MEDIA_ROOT).glob("**/*")
181 if not x.is_dir() and x.name not in settings.IGNORABLE_FILES
182 }
184 lockfile = Path(settings.MEDIA_LOCK).resolve()
185 present_files.discard(lockfile)
187 general_config = GeneralConfig()
188 app_logo = general_config.app_logo or settings.APP_LOGO
189 if app_logo:
190 logo_file = Path(settings.MEDIA_ROOT / Path(app_logo.lstrip("/"))).resolve()
191 present_files.discard(logo_file)
193 return present_files
196def _check_thumbnail(
197 doc: Document,
198 messages: SanityCheckMessages,
199 present_files: set[Path],
200) -> None:
201 """Verify the thumbnail exists and is readable."""
202 # doc.thumbnail_path already returns a resolved Path; no need to re-resolve.
203 thumbnail_path: Final[Path] = doc.thumbnail_path
204 if not thumbnail_path.is_file():
205 messages.error(doc.pk, "Thumbnail of document does not exist.")
206 return
208 present_files.discard(thumbnail_path)
209 try:
210 _ = thumbnail_path.read_bytes()
211 except OSError as e:
212 messages.error(doc.pk, f"Cannot read thumbnail file of document: {e}")
215def _check_original(
216 doc: Document,
217 messages: SanityCheckMessages,
218 present_files: set[Path],
219) -> None:
220 """Verify the original file exists, is readable, and has matching checksum."""
221 # doc.source_path already returns a resolved Path; no need to re-resolve.
222 source_path: Final[Path] = doc.source_path
223 if not source_path.is_file():
224 messages.error(doc.pk, "Original of document does not exist.")
225 return
227 present_files.discard(source_path)
228 try:
229 checksum = compute_checksum(source_path)
230 except OSError as e:
231 messages.error(doc.pk, f"Cannot read original file of document: {e}")
232 else:
233 if checksum != doc.checksum:
234 messages.error(
235 doc.pk,
236 f"Checksum mismatch. Stored: {doc.checksum}, actual: {checksum}.",
237 )
240def _check_archive(
241 doc: Document,
242 messages: SanityCheckMessages,
243 present_files: set[Path],
244) -> None:
245 """Verify archive file consistency: checksum/filename pairing and file integrity."""
246 if doc.archive_checksum is not None and doc.archive_filename is None:
247 messages.error(
248 doc.pk,
249 "Document has an archive file checksum, but no archive filename.",
250 )
251 elif doc.archive_checksum is None and doc.archive_filename is not None:
252 messages.error(
253 doc.pk,
254 "Document has an archive file, but its checksum is missing.",
255 )
256 elif doc.has_archive_version:
257 if TYPE_CHECKING:
258 assert isinstance(doc.archive_path, Path)
259 # doc.archive_path already returns a resolved Path; no need to re-resolve.
260 archive_path: Final[Path] = doc.archive_path # type: ignore[assignment]
261 if not archive_path.is_file():
262 messages.error(doc.pk, "Archived version of document does not exist.")
263 return
265 present_files.discard(archive_path)
266 try:
267 checksum = compute_checksum(archive_path)
268 except OSError as e:
269 messages.error(
270 doc.pk,
271 f"Cannot read archive file of document: {e}",
272 )
273 else:
274 if checksum != doc.archive_checksum:
275 messages.error(
276 doc.pk,
277 "Checksum mismatch of archived document. "
278 f"Stored: {doc.archive_checksum}, actual: {checksum}.",
279 )
282def _check_content(doc: Document, messages: SanityCheckMessages) -> None:
283 """Flag documents with no OCR content."""
284 if not doc.content:
285 messages.info(doc.pk, "Document contains no OCR data")
288def _check_document(
289 doc: Document,
290 messages: SanityCheckMessages,
291 present_files: set[Path],
292) -> None:
293 """Run all checks for a single document."""
294 _check_thumbnail(doc, messages, present_files)
295 _check_original(doc, messages, present_files)
296 _check_archive(doc, messages, present_files)
297 _check_content(doc, messages)
300# ---------------------------------------------------------------------------
301# Public entry point
302# ---------------------------------------------------------------------------
305def check_sanity(
306 *,
307 iter_wrapper: IterWrapper[Document] = identity,
308) -> SanityCheckMessages:
309 """Run a full sanity check on the document archive.
311 Args:
312 iter_wrapper: A callable that wraps the document iterable, e.g.,
313 for progress bar display. Defaults to identity (no wrapping).
315 Returns:
316 A SanityCheckMessages instance containing all detected issues.
317 """
318 messages = SanityCheckMessages()
319 present_files = _build_present_files()
321 documents = Document.global_objects.only(
322 "pk",
323 "filename",
324 "mime_type",
325 "checksum",
326 "archive_checksum",
327 "archive_filename",
328 "content",
329 ).iterator(chunk_size=500)
330 for doc in iter_wrapper(documents):
331 _check_document(doc, messages, present_files)
333 for extra_file in present_files:
334 messages.warning(None, f"Orphaned file in media dir: {extra_file}")
336 return messages