Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/parser_services/_base.py: 77%
107 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:32 +0000
1from abc import ABC, abstractmethod 1e
3from pydantic import UUID4, BaseModel 1e
4from rapidfuzz import fuzz, process 1e
5from sqlalchemy.orm import Session 1e
7from mealie.db.models.recipe.ingredient import IngredientFoodModel, IngredientUnitModel 1e
8from mealie.repos.all_repositories import get_repositories 1e
9from mealie.repos.repository_factory import AllRepositories 1e
10from mealie.schema.recipe.recipe_ingredient import ( 1e
11 CreateIngredientFood,
12 CreateIngredientUnit,
13 IngredientFood,
14 IngredientUnit,
15 ParsedIngredient,
16)
17from mealie.schema.response.pagination import PaginationQuery 1e
20class DataMatcher: 1e
21 def __init__( 1e
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 1ifjkhgdacb
29 self._food_fuzzy_match_threshold = food_fuzzy_match_threshold 1ifjkhgdacb
30 self._unit_fuzzy_match_threshold = unit_fuzzy_match_threshold 1ifjkhgdacb
31 self._foods_by_alias: dict[str, IngredientFood] | None = None 1ifjkhgdacb
32 self._units_by_alias: dict[str, IngredientUnit] | None = None 1ifjkhgdacb
34 @property 1e
35 def foods_by_alias(self) -> dict[str, IngredientFood]: 1e
36 if self._foods_by_alias is None: 1fhgdacb
37 foods_repo = self.repos.ingredient_foods 1fhgdacb
38 query = PaginationQuery(page=1, per_page=-1) 1fhgdacb
39 all_foods = foods_repo.page_all(query).items 1fhgdacb
41 foods_by_alias: dict[str, IngredientFood] = {} 1fhgdacb
42 for food in all_foods: 1fhgdacb
43 if food.name: 1fhgdacb
44 foods_by_alias[IngredientFoodModel.normalize(food.name)] = food 1fhgdacb
45 if food.plural_name: 1fhgdacb
46 foods_by_alias[IngredientFoodModel.normalize(food.plural_name)] = food 1fhgdacb
48 for alias in food.aliases or []: 1fhgdacb
49 if alias.name: 1fhgdacb
50 foods_by_alias[IngredientFoodModel.normalize(alias.name)] = food 1fhgdacb
52 self._foods_by_alias = foods_by_alias 1fhgdacb
54 return self._foods_by_alias 1fhgdacb
56 @property 1e
57 def units_by_alias(self) -> dict[str, IngredientUnit]: 1e
58 if self._units_by_alias is None: 1acb
59 units_repo = self.repos.ingredient_units 1acb
60 query = PaginationQuery(page=1, per_page=-1) 1acb
61 all_units = units_repo.page_all(query).items 1acb
63 units_by_alias: dict[str, IngredientUnit] = {} 1acb
64 for unit in all_units: 64 ↛ 65line 64 didn't jump to line 65 because the loop on line 64 never started1acb
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
74 for alias in unit.aliases or []:
75 if alias.name:
76 units_by_alias[IngredientUnitModel.normalize(alias.name)] = unit
78 self._units_by_alias = units_by_alias 1acb
80 return self._units_by_alias 1acb
82 @classmethod 1e
83 def find_match[T: BaseModel]( 1e
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: 1fhgdacb
88 return store_map[match_value] 1fhgacb
90 # fuzzy match against food store
91 fuzz_result = process.extractOne( 1fgdacb
92 match_value, store_map.keys(), scorer=fuzz.ratio, score_cutoff=fuzzy_match_threshold
93 )
94 if fuzz_result is None: 1fgdacb
95 return None 1fgdacb
97 return store_map[fuzz_result[0]] 1dab
99 def find_food_match(self, food: IngredientFood | CreateIngredientFood | str) -> IngredientFood | None: 1e
100 if isinstance(food, IngredientFood): 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true1fhgdacb
101 return food
103 food_name = food if isinstance(food, str) else food.name 1fhgdacb
104 match_value = IngredientFoodModel.normalize(food_name) 1fhgdacb
105 return self.find_match( 1fhgdacb
106 match_value,
107 store_map=self.foods_by_alias,
108 fuzzy_match_threshold=self._food_fuzzy_match_threshold,
109 )
111 def find_unit_match(self, unit: IngredientUnit | CreateIngredientUnit | str) -> IngredientUnit | None: 1e
112 if isinstance(unit, IngredientUnit): 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true1acb
113 return unit
115 unit_name = unit if isinstance(unit, str) else unit.name 1acb
116 match_value = IngredientUnitModel.normalize(unit_name) 1acb
117 return self.find_match( 1acb
118 match_value,
119 store_map=self.units_by_alias,
120 fuzzy_match_threshold=self._unit_fuzzy_match_threshold,
121 )
124class ABCIngredientParser(ABC): 1e
125 """
126 Abstract class for ingredient parsers.
127 """
129 def __init__(self, group_id: UUID4, session: Session) -> None: 1e
130 self.group_id = group_id 1dacb
131 self.session = session 1dacb
132 self.data_matcher = DataMatcher(self._repos, self.food_fuzzy_match_threshold, self.unit_fuzzy_match_threshold) 1dacb
134 @property 1e
135 def _repos(self) -> AllRepositories: 1e
136 return get_repositories(self.session, group_id=self.group_id) 1dacb
138 @property 1e
139 def food_fuzzy_match_threshold(self) -> int: 1e
140 """Minimum threshold to fuzzy match against a database food search"""
142 return 85 1dacb
144 @property 1e
145 def unit_fuzzy_match_threshold(self) -> int: 1e
146 """Minimum threshold to fuzzy match against a database unit search"""
148 return 70 1dacb
150 @abstractmethod 1e
151 async def parse_one(self, ingredient_string: str) -> ParsedIngredient: ... 151 ↛ exitline 151 didn't return from function 'parse_one' because 1e
153 @abstractmethod 1e
154 async def parse(self, ingredients: list[str]) -> list[ParsedIngredient]: ... 154 ↛ exitline 154 didn't return from function 'parse' because 1e
156 def find_ingredient_match(self, ingredient: ParsedIngredient) -> ParsedIngredient: 1e
157 if ingredient.ingredient.food and (food_match := self.data_matcher.find_food_match(ingredient.ingredient.food)): 1dacb
158 ingredient.ingredient.food = food_match 1dacb
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 true1dacb
161 ingredient.ingredient.unit = unit_match
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) 1dacb
165 unit_is_matched = bool(ingredient.ingredient.unit and ingredient.ingredient.unit.id) 1dacb
166 food_name = ingredient.ingredient.food.name if ingredient.ingredient.food else "" 1dacb
167 unit_name = ingredient.ingredient.unit.name if ingredient.ingredient.unit else "" 1dacb
169 if not food_is_matched and not unit_is_matched and food_name and unit_name: 1dacb
170 food_match = self.data_matcher.find_food_match(f"{unit_name} {food_name}") 1acb
172 if food_match: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true1acb
173 ingredient.ingredient.food = food_match
174 ingredient.ingredient.unit = None
176 return ingredient 1dacb