Coverage for src/lilbee/wiki/wipe.py: 100%

1"""Remove a generated wiki: its pages on disk and its rows in the store.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import shutil 

7from dataclasses import dataclass 

8from typing import TYPE_CHECKING 

9 

10from lilbee.core.config import cfg 

11 

12from .shared import WIKI_BUILD_LOCK, WIKI_CONTENT_SUBDIRS, WikiSubdir, count_pages_in 

13 

14if TYPE_CHECKING: 

15 from lilbee.core.config import Config 

16 from lilbee.data.store import Store 

17 

18log = logging.getLogger(__name__) 

19 

20# Every subdir holding pages a wipe removes: published content plus the 

21# quarantined drafts and the archive prune leaves behind. 

22_PAGE_SUBDIRS: tuple[WikiSubdir, ...] = ( 

23 *WIKI_CONTENT_SUBDIRS, 

24 WikiSubdir.DRAFTS, 

25 WikiSubdir.ARCHIVE, 

26) 

27 

28 

29@dataclass(frozen=True) 

30class WipeReport: 

31 """What a wipe removed, and whether the store delete actually landed.""" 

32 

33 pages_removed: int 

34 sources_cleared: int 

35 rows_deleted: bool 

36 

37 def summary(self) -> str: 

38 """One line for the CLI, MCP, and HTTP responses.""" 

39 pages = f"{self.pages_removed} page{'s' if self.pages_removed != 1 else ''}" 

40 if not self.rows_deleted: 

41 return f"Removed {pages}, but deleting the store rows failed; run the wipe again" 

42 return f"Removed {pages} and the store rows for {self.sources_cleared} of them" 

43 

44 

45def wipe_wiki(store: Store, config: Config | None = None) -> WipeReport: 

46 """Delete every generated wiki page and the store rows behind it. 

47 

48 Pages go first. A crash in between then leaves rows whose page is gone, 

49 which the next prune reconciles away; the reverse order would leave pages 

50 on disk that no check ever retries, because an uncited page reads as 

51 nothing to do. 

52 """ 

53 if config is None: 

54 config = cfg 

55 wiki_root = config.data_root / config.wiki_dir 

56 with WIKI_BUILD_LOCK: 

57 sources = store.wiki_chunk_sources() | store.wiki_citation_sources() 

58 pages_removed = count_pages_in(wiki_root, _PAGE_SUBDIRS) 

59 if wiki_root.is_dir(): 

60 shutil.rmtree(wiki_root) 

61 rows_deleted = store.delete_all_wiki_rows() 

62 if not rows_deleted: 

63 log.warning("Wiki wipe removed the pages but failed to delete the store rows") 

64 log.info("Wiki wipe: %d pages, %d indexed sources", pages_removed, len(sources)) 

65 return WipeReport( 

66 pages_removed=pages_removed, 

67 sources_cleared=len(sources), 

68 rows_deleted=rows_deleted, 

69 )