munch-ease-backend/recipes/scraping.py

from bs4 import BeautifulSoup
import httpx
import json

def _is_recipe_ldata(ldata_node):
    if '@type' in ldata_node:
        typ = ldata_node['@type']
        if isinstance(typ, list):
            typ = typ[0]

        if isinstance(typ, str) and typ.lower() == 'recipe':
            return True
    
    return None

async def scrape_recipe(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 _is_recipe_ldata(data):
                    return data
                
                if '@graph' in data:
                    for item in data['@graph']:
                        if _is_recipe_ldata(item):
                            return item
                        
                if isinstance(data, list):
                    for item in data:
                        if _is_recipe_ldata(item):
                            return item

            except (json.decoder.JSONDecodeError, KeyError):
                pass
        
        return None

def _dump_json_data_to_log(data: dict) -> str:
    import os, re
    dir = './data/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)