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-11-25 17:29 +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 1fghdabijklmnopqrstuvwxyzABCDEFc
29 self._food_fuzzy_match_threshold = food_fuzzy_match_threshold 1fghdabijklmnopqrstuvwxyzABCDEFc
30 self._unit_fuzzy_match_threshold = unit_fuzzy_match_threshold 1fghdabijklmnopqrstuvwxyzABCDEFc
31 self._foods_by_alias: dict[str, IngredientFood] | None = None 1fghdabijklmnopqrstuvwxyzABCDEFc
32 self._units_by_alias: dict[str, IngredientUnit] | None = None 1fghdabijklmnopqrstuvwxyzABCDEFc
34 @property 1e
35 def foods_by_alias(self) -> dict[str, IngredientFood]: 1e
36 if self._foods_by_alias is None: 1dabc
37 foods_repo = self.repos.ingredient_foods 1dabc
38 query = PaginationQuery(page=1, per_page=-1) 1dabc
39 all_foods = foods_repo.page_all(query).items 1dabc
41 foods_by_alias: dict[str, IngredientFood] = {} 1dabc
42 for food in all_foods: 1dabc
43 if food.name: 1dabc
44 foods_by_alias[IngredientFoodModel.normalize(food.name)] = food 1dabc
45 if food.plural_name: 1dabc
46 foods_by_alias[IngredientFoodModel.normalize(food.plural_name)] = food 1dabc
48 for alias in food.aliases or []: 1dabc
49 if alias.name: 1dabc
50 foods_by_alias[IngredientFoodModel.normalize(alias.name)] = food 1dabc
52 self._foods_by_alias = foods_by_alias 1dabc
54 return self._foods_by_alias 1dabc
56 @property 1e
57 def units_by_alias(self) -> dict[str, IngredientUnit]: 1e
58 if self._units_by_alias is None: 1abc
59 units_repo = self.repos.ingredient_units 1abc
60 query = PaginationQuery(page=1, per_page=-1) 1abc
61 all_units = units_repo.page_all(query).items 1abc
63 units_by_alias: dict[str, IngredientUnit] = {} 1abc
64 for unit in all_units: 64 ↛ 65line 64 didn't jump to line 65 because the loop on line 64 never started1abc
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 1abc
80 return self._units_by_alias 1abc
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: 1dabc
88 return store_map[match_value] 1dabc
90 # fuzzy match against food store
91 fuzz_result = process.extractOne( 1dabc
92 match_value, store_map.keys(), scorer=fuzz.ratio, score_cutoff=fuzzy_match_threshold
93 )
94 if fuzz_result is None: 1dabc
95 return None 1dabc
97 return store_map[fuzz_result[0]] 1abc
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 true1dabc
101 return food
103 food_name = food if isinstance(food, str) else food.name 1dabc
104 match_value = IngredientFoodModel.normalize(food_name) 1dabc
105 return self.find_match( 1dabc
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 true1abc
113 return unit
115 unit_name = unit if isinstance(unit, str) else unit.name 1abc
116 match_value = IngredientUnitModel.normalize(unit_name) 1abc
117 return self.find_match( 1abc
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 1dabc
131 self.session = session 1dabc
132 self.data_matcher = DataMatcher(self._repos, self.food_fuzzy_match_threshold, self.unit_fuzzy_match_threshold) 1dabc
134 @property 1e
135 def _repos(self) -> AllRepositories: 1e
136 return get_repositories(self.session, group_id=self.group_id) 1dabc
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 1dabc
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 1dabc
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)): 1dabc
158 ingredient.ingredient.food = food_match 1dabc
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 true1dabc
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) 1dabc
165 unit_is_matched = bool(ingredient.ingredient.unit and ingredient.ingredient.unit.id) 1dabc
166 food_name = ingredient.ingredient.food.name if ingredient.ingredient.food else "" 1dabc
167 unit_name = ingredient.ingredient.unit.name if ingredient.ingredient.unit else "" 1dabc
169 if not food_is_matched and not unit_is_matched and food_name and unit_name: 1dabc
170 food_match = self.data_matcher.find_food_match(f"{unit_name} {food_name}") 1abc
172 if food_match: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true1abc
173 ingredient.ingredient.food = food_match
174 ingredient.ingredient.unit = None
176 return ingredient 1dabc