Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/parser_services/_base.py: 72%

1from abc import ABC, abstractmethod 1 ctx1b

2 

3from pydantic import UUID4, BaseModel 1 ctx1b

4from rapidfuzz import fuzz, process 1 ctx1b

5from sqlalchemy.orm import Session 1 ctx1b

6 

7from mealie.db.models.recipe.ingredient import IngredientFoodModel, IngredientUnitModel 1 ctx1b

8from mealie.repos.all_repositories import get_repositories 1 ctx1b

9from mealie.repos.repository_factory import AllRepositories 1 ctx1b

10from mealie.schema.recipe.recipe_ingredient import ( 1 ctx1b

11 CreateIngredientFood, 

12 CreateIngredientUnit, 

13 IngredientFood, 

14 IngredientUnit, 

15 ParsedIngredient, 

16) 

17from mealie.schema.response.pagination import PaginationQuery 1 ctx1b

18 

19 

20class DataMatcher: 1 ctx1b

21 def __init__( 1 ctx1b

22 self, 

23 repos: AllRepositories, 

24 food_fuzzy_match_threshold: int = 85, 

25 unit_fuzzy_match_threshold: int = 70, 

26 ) -> None: 

27 self.repos = repos 17 ctx1cdefghijklmnopqra

28 

29 self._food_fuzzy_match_threshold = food_fuzzy_match_threshold 17 ctx1cdefghijklmnopqra

30 self._unit_fuzzy_match_threshold = unit_fuzzy_match_threshold 17 ctx1cdefghijklmnopqra

31 self._foods_by_alias: dict[str, IngredientFood] | None = None 17 ctx1cdefghijklmnopqra

32 self._units_by_alias: dict[str, IngredientUnit] | None = None 17 ctx1cdefghijklmnopqra

33 

34 @property 1 ctx1b

35 def foods_by_alias(self) -> dict[str, IngredientFood]: 1 ctx1b

36 if self._foods_by_alias is None: (empty)

37 foods_repo = self.repos.ingredient_foods (empty)

38 query = PaginationQuery(page=1, per_page=-1) (empty)

39 all_foods = foods_repo.page_all(query).items (empty)

40 

41 foods_by_alias: dict[str, IngredientFood] = {} (empty)

42 for food in all_foods: (empty)

43 if food.name: (empty)

44 foods_by_alias[IngredientFoodModel.normalize(food.name)] = food (empty)

45 if food.plural_name: (empty)

46 foods_by_alias[IngredientFoodModel.normalize(food.plural_name)] = food (empty)

47 

48 for alias in food.aliases or []: 48 ↛ 49line 48 didn't jump to line 49 because the loop on line 48 never started(empty)

49 if alias.name: 

50 foods_by_alias[IngredientFoodModel.normalize(alias.name)] = food 

51 

52 self._foods_by_alias = foods_by_alias (empty)

53 

54 return self._foods_by_alias (empty)

55 

56 @property 1 ctx1b

57 def units_by_alias(self) -> dict[str, IngredientUnit]: 1 ctx1b

58 if self._units_by_alias is None: (empty)

59 units_repo = self.repos.ingredient_units (empty)

60 query = PaginationQuery(page=1, per_page=-1) (empty)

61 all_units = units_repo.page_all(query).items (empty)

62 

63 units_by_alias: dict[str, IngredientUnit] = {} (empty)

64 for unit in all_units: 64 ↛ 65line 64 didn't jump to line 65 because the loop on line 64 never started(empty)

65 if unit.name: 

66 units_by_alias[IngredientUnitModel.normalize(unit.name)] = unit 

67 if unit.plural_name: 

68 units_by_alias[IngredientUnitModel.normalize(unit.plural_name)] = unit 

69 if unit.abbreviation: 

70 units_by_alias[IngredientUnitModel.normalize(unit.abbreviation)] = unit 

71 if unit.plural_abbreviation: 

72 units_by_alias[IngredientUnitModel.normalize(unit.plural_abbreviation)] = unit 

73 

74 for alias in unit.aliases or []: 

75 if alias.name: 

76 units_by_alias[IngredientUnitModel.normalize(alias.name)] = unit 

77 

78 self._units_by_alias = units_by_alias (empty)

79 

80 return self._units_by_alias (empty)

81 

82 @classmethod 1 ctx1b

83 def find_match[T: BaseModel]( 1 ctx1b

84 cls, match_value: str, *, store_map: dict[str, T], fuzzy_match_threshold: int = 0 

85 ) -> T | None: 

86 # check for literal matches 

87 if match_value in store_map: (empty)

88 return store_map[match_value] (empty)

89 

90 # fuzzy match against food store 

91 fuzz_result = process.extractOne( (empty)

92 match_value, store_map.keys(), scorer=fuzz.ratio, score_cutoff=fuzzy_match_threshold 

93 ) 

94 if fuzz_result is None: (empty)

95 return None (empty)

96 

97 return store_map[fuzz_result[0]] (empty)

98 

99 def find_food_match(self, food: IngredientFood | CreateIngredientFood | str) -> IngredientFood | None: 1 ctx1b

100 if isinstance(food, IngredientFood): 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true(empty)

101 return food 

102 

103 food_name = food if isinstance(food, str) else food.name (empty)

104 match_value = IngredientFoodModel.normalize(food_name) (empty)

105 return self.find_match( (empty)

106 match_value, 

107 store_map=self.foods_by_alias, 

108 fuzzy_match_threshold=self._food_fuzzy_match_threshold, 

109 ) 

110 

111 def find_unit_match(self, unit: IngredientUnit | CreateIngredientUnit | str) -> IngredientUnit | None: 1 ctx1b

112 if isinstance(unit, IngredientUnit): 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true(empty)

113 return unit 

114 

115 unit_name = unit if isinstance(unit, str) else unit.name (empty)

116 match_value = IngredientUnitModel.normalize(unit_name) (empty)

117 return self.find_match( (empty)

118 match_value, 

119 store_map=self.units_by_alias, 

120 fuzzy_match_threshold=self._unit_fuzzy_match_threshold, 

121 ) 

122 

123 

124class ABCIngredientParser(ABC): 1 ctx1b

125 """ 

126 Abstract class for ingredient parsers. 

127 """ 

128 

129 def __init__(self, group_id: UUID4, session: Session) -> None: 1 ctx1b

130 self.group_id = group_id (empty)

131 self.session = session (empty)

132 self.data_matcher = DataMatcher(self._repos, self.food_fuzzy_match_threshold, self.unit_fuzzy_match_threshold) (empty)

133 

134 @property 1 ctx1b

135 def _repos(self) -> AllRepositories: 1 ctx1b

136 return get_repositories(self.session, group_id=self.group_id) (empty)

137 

138 @property 1 ctx1b

139 def food_fuzzy_match_threshold(self) -> int: 1 ctx1b

140 """Minimum threshold to fuzzy match against a database food search""" 

141 

142 return 85 (empty)

143 

144 @property 1 ctx1b

145 def unit_fuzzy_match_threshold(self) -> int: 1 ctx1b

146 """Minimum threshold to fuzzy match against a database unit search""" 

147 

148 return 70 (empty)

149 

150 @abstractmethod 1 ctx1b

151 async def parse_one(self, ingredient_string: str) -> ParsedIngredient: ... 151 ↛ exitline 151 didn't return from function 'parse_one' because 1 ctx1b

152 

153 @abstractmethod 1 ctx1b

154 async def parse(self, ingredients: list[str]) -> list[ParsedIngredient]: ... 154 ↛ exitline 154 didn't return from function 'parse' because 1 ctx1b

155 

156 def find_ingredient_match(self, ingredient: ParsedIngredient) -> ParsedIngredient: 1 ctx1b

157 if ingredient.ingredient.food and (food_match := self.data_matcher.find_food_match(ingredient.ingredient.food)): (empty)

158 ingredient.ingredient.food = food_match (empty)

159 

160 if ingredient.ingredient.unit and (unit_match := self.data_matcher.find_unit_match(ingredient.ingredient.unit)): 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true(empty)

161 ingredient.ingredient.unit = unit_match 

162 

163 # Parser might have wrongly split a food into a unit and food. 

164 food_is_matched = bool(ingredient.ingredient.food and ingredient.ingredient.food.id) (empty)

165 unit_is_matched = bool(ingredient.ingredient.unit and ingredient.ingredient.unit.id) (empty)

166 food_name = ingredient.ingredient.food.name if ingredient.ingredient.food else "" (empty)

167 unit_name = ingredient.ingredient.unit.name if ingredient.ingredient.unit else "" (empty)

168 

169 if not food_is_matched and not unit_is_matched and food_name and unit_name: 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true(empty)

170 food_match = self.data_matcher.find_food_match(f"{unit_name} {food_name}") 

171 

172 if food_match: 

173 ingredient.ingredient.food = food_match 

174 ingredient.ingredient.unit = None 

175 

176 return ingredient (empty)