munch-ease-backend/ingredients/__init__.py

from ingredients.db import Ingredient, find_ingredients_by_meal_id, find_ingredients_by_recipe_id, insert_ingredient, delete_ingredients_by_meal_id

import units
from products import Product, find_product_by_tag, get_or_create, add_missing_tags

from ingredient_parser import parse_ingredient

import re
from typing import List

async def parse_ingredient_from_link(conn, link: str) -> 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
        )
    
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) -> Product:
    gen = find_product_by_tag(conn, ingredient)
    try:
        async for item in gen:
            return item
    finally:
        await gen.aclose()

    return 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