import re from typing import List, Optional from ingredient_parser import parse_ingredient import units from ingredients.db import ( Ingredient, delete_ingredients_by_meal_id as delete_ingredients_by_meal_id, find_ingredient_by_id as find_ingredient_by_id, find_ingredients_by_meal_id as find_ingredients_by_meal_id, find_ingredients_by_recipe_id as find_ingredients_by_recipe_id, find_ingredients_by_recipe_ids as find_ingredients_by_recipe_ids, insert_ingredient as insert_ingredient, ) from products import Product, add_missing_tags, find_product_by_tag, get_or_create async def parse_ingredient_from_link(conn, link: str) -> Optional[Ingredient]: match = re.match(r"^(\d+)?\s*(http.*)$", link) if not match: return None quantity = int(match.group(1)) if match.group(1) else 1 url = match.group(2) product = await get_or_create(conn, url, []) if product: await add_missing_tags(conn, product, [product.name]) return Ingredient( id=-1, name=product.name, line=f"{quantity}x {product.name}", unit=units.ITEMS.name, quantity=quantity, preparation="", product_id=product.id, product=product, ) return None def parse_ingredient_from_nlp(ingredient_string: str) -> Ingredient: ingredient = parse_ingredient(ingredient_string) name = ingredient.name.text if ingredient.name else "" quantity, unit = None, None for amount in ingredient.amount: if quantity is None and amount.quantity: quantity = amount.quantity if unit is None and amount.unit: real_unit = units.get_unit(str(amount.unit)) if real_unit: unit = real_unit.name if isinstance(quantity, str): try: quantity = float(quantity) except ValueError: pass if quantity is None or not isinstance(quantity, (int, float)): quantity = 1 if unit is None: unit = units.ITEMS.name return Ingredient( id=-1, line=ingredient.sentence, name=name, quantity=quantity, unit=unit, preparation=ingredient.preparation.text if ingredient.preparation else "", product_id=-1, ) async def _find_existing_product(conn, ingredient: str) -> Optional[Product]: items = [item async for item in find_product_by_tag(conn, ingredient)] return items[0] if items else None async def match_existing_products(conn, ingredients: List[Ingredient]) -> List[Ingredient]: for ingredient in ingredients: if not ingredient.product: existing = await _find_existing_product(conn, ingredient.name) if existing: ingredient.product_id = existing.id ingredient.product = existing return ingredients