110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
import re
|
|
from typing import Optional
|
|
|
|
from ingredients import match_existing_products, parse_ingredient_from_nlp
|
|
from api.dtos import MemberRef
|
|
from recipes.models import Recipe as Recipe
|
|
from recipes.repository import (
|
|
compute_prev_cursor as compute_prev_cursor,
|
|
count_all as count_all,
|
|
count_by_name as count_by_name,
|
|
count_all_scoped as count_all_scoped,
|
|
count_by_name_scoped as count_by_name_scoped,
|
|
find_recipe_by_id as find_recipe_by_id,
|
|
find_recipe_by_id_scoped as find_recipe_by_id_scoped,
|
|
find_recipes_by_name as find_recipes_by_name,
|
|
find_recipes_by_name_paged as find_recipes_by_name_paged,
|
|
find_recipes_by_name_paged_scoped as find_recipes_by_name_paged_scoped,
|
|
get_all as get_all,
|
|
get_all_paged as get_all_paged,
|
|
get_all_paged_scoped as get_all_paged_scoped,
|
|
insert_recipe as insert_recipe,
|
|
insert_recipe_scoped as insert_recipe_scoped,
|
|
load_recipe_ingredients as load_recipe_ingredients,
|
|
row_to_recipe as row_to_recipe,
|
|
)
|
|
from recipes.scraping import (
|
|
scrape_recipe_ldata as _scrape_recipe_ldata,
|
|
scrape_recipe_ldata_from_html as _scrape_recipe_ldata_from_html,
|
|
)
|
|
|
|
|
|
async def parse_recipe(conn, created_by, url: str, log=None, dump_dir: str | None = None) -> Optional[Recipe]:
|
|
"""Parse a recipe from a URL. Returns None if parsing fails.
|
|
|
|
Accepts an optional log callable taking a single string argument; when provided,
|
|
the scraper will emit diagnostic messages useful for manual testing.
|
|
"""
|
|
ldata = await _scrape_recipe_ldata(url, log=log, dump_dir=dump_dir)
|
|
|
|
if ldata:
|
|
return await _get_recipe_from_ldata(conn, url, ldata, created_by)
|
|
return None
|
|
|
|
|
|
async def parse_recipe_from_html(conn, created_by, base_url: str, html: str, log=None) -> Optional[Recipe]:
|
|
"""Parse a recipe from raw HTML (offline). Returns None if parsing fails.
|
|
|
|
This mirrors parse_recipe() but uses already-downloaded HTML via the offline
|
|
scraper entrypoint. Useful for tests/assertions against saved snapshots.
|
|
"""
|
|
ldata = _scrape_recipe_ldata_from_html(html, base_url, log=log)
|
|
if ldata:
|
|
return await _get_recipe_from_ldata(conn, base_url, ldata, created_by)
|
|
return None
|
|
|
|
|
|
def find_yield(recipe_ldata: dict) -> int:
|
|
if "recipeYield" in recipe_ldata:
|
|
yield_vals = recipe_ldata["recipeYield"]
|
|
if not isinstance(yield_vals, list):
|
|
yield_vals = [yield_vals]
|
|
|
|
for val in yield_vals:
|
|
try:
|
|
return int(val)
|
|
except ValueError:
|
|
pass
|
|
|
|
for val in yield_vals:
|
|
match = re.match(r"(\d+)", val)
|
|
if match:
|
|
return int(match.group(1))
|
|
|
|
return 4
|
|
|
|
|
|
async def _get_recipe_from_ldata(conn, url: str, ldata: dict, created_by) -> Recipe:
|
|
ingredients = [
|
|
parse_ingredient_from_nlp(ingredient) for ingredient in ldata["recipeIngredient"]
|
|
]
|
|
ingredients = await match_existing_products(conn, ingredients)
|
|
name = ldata["name"] if "name" in ldata else url
|
|
images = ldata["image"] if "image" in ldata else []
|
|
serves = find_yield(ldata)
|
|
|
|
if isinstance(images, list) and len(images) > 0 and isinstance(images[0], dict):
|
|
images = [image["url"] for image in images]
|
|
|
|
if isinstance(images, dict):
|
|
images = [images["url"]]
|
|
|
|
if isinstance(images, str):
|
|
images = [images]
|
|
|
|
# Ensure created_by is a MemberRef (not a full User) to satisfy model typing
|
|
mref = (
|
|
MemberRef(id=created_by.id, display_name=created_by.display_name)
|
|
if created_by is not None
|
|
else None
|
|
)
|
|
return Recipe(
|
|
id=-1,
|
|
name=name,
|
|
link=url,
|
|
serves=serves,
|
|
image_urls=images,
|
|
ingredients=ingredients,
|
|
created_by=mref,
|
|
created_by_id=created_by.id if created_by is not None else -1,
|
|
)
|