Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/admin/admin_maintenance.py: 84%

1import shutil 

2import uuid 

3from pathlib import Path 

4 

5from fastapi import APIRouter, HTTPException 

6 

7from mealie.pkgs.stats import fs_stats 

8from mealie.routes._base import BaseAdminController, controller 

9from mealie.schema.admin import MaintenanceSummary 

10from mealie.schema.admin.maintenance import MaintenanceStorageDetails 

11from mealie.schema.response import ErrorResponse, SuccessResponse 

12 

13router = APIRouter(prefix="/maintenance") 

14 

15 

16def clean_images(root_dir: Path, dry_run: bool) -> int: 

17 cleaned_images = 0 

18 

19 for recipe_dir in root_dir.iterdir(): 

20 image_dir = recipe_dir.joinpath("images") 

21 

22 if not image_dir.exists(): 

23 continue 

24 

25 for image in image_dir.iterdir(): 

26 if image.is_dir(): 

27 continue 

28 

29 if image.suffix != ".webp": 

30 if not dry_run: 30 ↛ 33line 30 didn't jump to line 33 because the condition on line 30 was always true

31 image.unlink() 

32 

33 cleaned_images += 1 

34 

35 return cleaned_images 

36 

37 

38def clean_recipe_folders(root_dir: Path, dry_run: bool) -> int: 

39 cleaned_dirs = 0 

40 

41 for recipe_dir in root_dir.iterdir(): 

42 if recipe_dir.is_dir(): 42 ↛ 41line 42 didn't jump to line 41 because the condition on line 42 was always true

43 # Attempt to convert the folder name to a UUID 

44 try: 

45 uuid.UUID(recipe_dir.name) 

46 continue 

47 except ValueError: 

48 if not dry_run: 48 ↛ 50line 48 didn't jump to line 50 because the condition on line 48 was always true

49 shutil.rmtree(recipe_dir) 

50 cleaned_dirs += 1 

51 

52 return cleaned_dirs 

53 

54 

55def tail_log(log_file: Path, n: int) -> list[str]: 

56 try: 

57 with open(log_file) as f: 

58 lines = f.readlines() 

59 except FileNotFoundError: 

60 return ["no log file found"] 

61 

62 return lines[-n:] 

63 

64 

65@controller(router) 

66class AdminMaintenanceController(BaseAdminController): 

67 @router.get("", response_model=MaintenanceSummary) 

68 def get_maintenance_summary(self): 

69 """ 

70 Get the maintenance summary 

71 """ 

72 

73 return MaintenanceSummary( 

74 data_dir_size=fs_stats.pretty_size(fs_stats.get_dir_size(self.folders.DATA_DIR)), 

75 cleanable_images=clean_images(self.folders.RECIPE_DATA_DIR, dry_run=True), 

76 cleanable_dirs=clean_recipe_folders(self.folders.RECIPE_DATA_DIR, dry_run=True), 

77 ) 

78 

79 @router.get("/storage", response_model=MaintenanceStorageDetails) 

80 def get_storage_details(self): 

81 return MaintenanceStorageDetails( 

82 temp_dir_size=fs_stats.pretty_size(fs_stats.get_dir_size(self.folders.TEMP_DIR)), 

83 backups_dir_size=fs_stats.pretty_size(fs_stats.get_dir_size(self.folders.BACKUP_DIR)), 

84 groups_dir_size=fs_stats.pretty_size(fs_stats.get_dir_size(self.folders.GROUPS_DIR)), 

85 recipes_dir_size=fs_stats.pretty_size(fs_stats.get_dir_size(self.folders.RECIPE_DATA_DIR)), 

86 user_dir_size=fs_stats.pretty_size(fs_stats.get_dir_size(self.folders.USER_DIR)), 

87 ) 

88 

89 @router.post("/clean/images", response_model=SuccessResponse) 

90 def clean_images(self): 

91 """ 

92 Purges all the images from the filesystem that aren't .webp 

93 """ 

94 try: 

95 cleaned_images = clean_images(self.folders.RECIPE_DATA_DIR, dry_run=False) 

96 return SuccessResponse.respond(f"{cleaned_images} Images cleaned") 

97 except Exception as e: 

98 raise HTTPException(status_code=500, detail=ErrorResponse.respond("Failed to clean images")) from e 

99 

100 @router.post("/clean/temp", response_model=SuccessResponse) 

101 def clean_temp(self): 

102 try: 

103 if self.folders.TEMP_DIR.exists(): 

104 shutil.rmtree(self.folders.TEMP_DIR) 

105 

106 self.folders.TEMP_DIR.mkdir(parents=True, exist_ok=True) 

107 except Exception as e: 

108 raise HTTPException(status_code=500, detail=ErrorResponse.respond("Failed to clean temp")) from e 

109 

110 return SuccessResponse.respond("'.temp' directory cleaned") 

111 

112 @router.post("/clean/recipe-folders", response_model=SuccessResponse) 

113 def clean_recipe_folders(self): 

114 """ 

115 Deletes all the recipe folders that don't have names that are valid UUIDs 

116 """ 

117 try: 

118 cleaned_dirs = clean_recipe_folders(self.folders.RECIPE_DATA_DIR, dry_run=False) 

119 return SuccessResponse.respond(f"{cleaned_dirs} Recipe folders removed") 

120 except Exception as e: 

121 raise HTTPException(status_code=500, detail=ErrorResponse.respond("Failed to clean directories")) from e