97 lines
No EOL
3.5 KiB
Python
97 lines
No EOL
3.5 KiB
Python
from model import Product, Ingredient, Recipe, Measurement
|
|
from ingredient_parser import parse_multiple_ingredients
|
|
from typing import List
|
|
from bs4 import BeautifulSoup
|
|
|
|
import httpx
|
|
import json
|
|
import product.db as db
|
|
|
|
async def find_existing_product(conn, ingredient: str) -> Product:
|
|
async for item in db.find_product_by_tag(conn, ingredient):
|
|
return item
|
|
return None
|
|
|
|
def parse_ingredient_from_nlp(ingredients: List[str]) -> Ingredient:
|
|
results = []
|
|
for ingredient in parse_multiple_ingredients(ingredients):
|
|
name = ingredient.name.text if ingredient.name else ''
|
|
|
|
if ingredient.amount:
|
|
measure = Measurement(ingredient.amount[0].quantity, ingredient.amount[0].unit)
|
|
else:
|
|
measure = Measurement(0, '')
|
|
|
|
results.append(Ingredient(
|
|
id=0,
|
|
line=ingredient.sentence,
|
|
name=name,
|
|
product=None,
|
|
measure=measure,
|
|
preparation=ingredient.preparation.text if ingredient.preparation else ''
|
|
))
|
|
|
|
return results
|
|
|
|
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 = existing
|
|
return ingredients
|
|
|
|
async def get_recipe_from_ldata(conn, url: str, ldata: dict) -> dict:
|
|
ingredients = parse_ingredient_from_nlp(ldata['recipeIngredient'])
|
|
ingredients = await match_existing_products(conn, ingredients)
|
|
return Recipe(
|
|
id=0,
|
|
name=ldata['name'],
|
|
link=url,
|
|
ingredients=ingredients
|
|
)
|
|
|
|
async def parse_recipe(conn, url: str) -> dict:
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
|
|
"Accept-Language": "en-US,en;q=0.9",
|
|
"Referer": "https://www.google.com/",
|
|
}
|
|
|
|
# Load the requested URL with headers
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(url, headers=headers, follow_redirects=True)
|
|
response.raise_for_status()
|
|
|
|
# Extract the recipe ld+json data
|
|
soup = BeautifulSoup(response.text, 'html.parser')
|
|
for ld in soup.find_all('script', type='application/ld+json'):
|
|
try:
|
|
data = json.loads(ld.text)
|
|
#_dump_json_data_to_log(data)
|
|
if '@type' in data and data['@type'].lower() == 'recipe':
|
|
return { 'raw': data, 'recipe': await get_recipe_from_ldata(conn, url, data) }
|
|
|
|
if '@graph' in data:
|
|
for item in data['@graph']:
|
|
if '@type' in item and item['@type'].lower() == 'recipe':
|
|
return { 'raw': data, 'recipe': await get_recipe_from_ldata(conn, url, item) }
|
|
|
|
except (json.decoder.JSONDecodeError, KeyError):
|
|
pass
|
|
|
|
return None
|
|
|
|
def _dump_json_data_to_log(data: dict) -> str:
|
|
import os, re
|
|
dir = './dump'
|
|
if not os.path.exists(dir):
|
|
os.makedirs(dir)
|
|
|
|
prefix = 'ldata_'
|
|
suffix = '.json'
|
|
file_ids = [int(re.findall(r'\d+', f)[0]) for f in os.listdir(dir) if re.match(prefix + r'\d+' + suffix, f)]
|
|
id = max(file_ids) + 1 if file_ids else 0
|
|
filename = f'{prefix}{id}{suffix}'
|
|
with open(os.path.join(dir, filename), 'w') as f:
|
|
json.dump(data, f, indent=4) |