92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
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
|