munch-ease-backend/ingredients/__init__.py

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,
    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