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

1from abc import ABC, abstractmethod 

2 

3from pydantic import UUID4, BaseModel 

4from rapidfuzz import fuzz, process 

5from sqlalchemy.orm import Session 

6 

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

8from mealie.repos.all_repositories import get_repositories 

9from mealie.repos.repository_factory import AllRepositories 

10from mealie.schema.recipe.recipe_ingredient import ( 

11 CreateIngredientFood, 

12 CreateIngredientUnit, 

13 IngredientFood, 

14 IngredientUnit, 

15 ParsedIngredient, 

16) 

17from mealie.schema.response.pagination import PaginationQuery 

18 

19 

20class DataMatcher: 

21 def __init__( 

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 

28 

29 self._food_fuzzy_match_threshold = food_fuzzy_match_threshold 

30 self._unit_fuzzy_match_threshold = unit_fuzzy_match_threshold 

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

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

33 

34 @property 

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

36 if self._foods_by_alias is None: 

37 foods_repo = self.repos.ingredient_foods 

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

39 all_foods = foods_repo.page_all(query).items 

40 

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

42 for food in all_foods: 

43 if food.name: 

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

45 if food.plural_name: 

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

47 

48 for alias in food.aliases or []: 

49 if alias.name: 

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

51 

52 self._foods_by_alias = foods_by_alias 

53 

54 return self._foods_by_alias 

55 

56 @property 

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

58 if self._units_by_alias is None: 

59 units_repo = self.repos.ingredient_units 

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

61 all_units = units_repo.page_all(query).items 

62 

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

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

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 

79 

80 return self._units_by_alias 

81 

82 @classmethod 

83 def find_match[T: BaseModel]( 

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: 

88 return store_map[match_value] 

89 

90 # fuzzy match against food store 

91 fuzz_result = process.extractOne( 

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

93 ) 

94 if fuzz_result is None: 

95 return None 

96 

97 return store_map[fuzz_result[0]] 

98 

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

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

101 return food 

102 

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

104 match_value = IngredientFoodModel.normalize(food_name) 

105 return self.find_match( 

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: 

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

113 return unit 

114 

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

116 match_value = IngredientUnitModel.normalize(unit_name) 

117 return self.find_match( 

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): 

125 """ 

126 Abstract class for ingredient parsers. 

127 """ 

128 

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

130 self.group_id = group_id 

131 self.session = session 

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

133 

134 @property 

135 def _repos(self) -> AllRepositories: 

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

137 

138 @property 

139 def food_fuzzy_match_threshold(self) -> int: 

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

141 

142 return 85 

143 

144 @property 

145 def unit_fuzzy_match_threshold(self) -> int: 

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

147 

148 return 70 

149 

150 @abstractmethod 

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

152 

153 @abstractmethod 

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

155 

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

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

158 ingredient.ingredient.food = food_match 

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

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) 

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

166 food_name = ingredient.ingredient.food.name if ingredient.ingredient.food else "" 

167 unit_name = ingredient.ingredient.unit.name if ingredient.ingredient.unit else "" 

168 

169 if not food_is_matched and not unit_is_matched and food_name and unit_name: 

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

171 

172 if food_match: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true

173 ingredient.ingredient.food = food_match 

174 ingredient.ingredient.unit = None 

175 

176 return ingredient