scraper.py

import argparse
import json
import logging
import math
import os
import secrets
import webbrowser
import zlib

import requests
import tqdm.contrib.concurrent


# Shamelessly stolen from: https://stackoverflow.com/a/69478073
def remove_empty_elements(jsonData):
    to_be_deleted = [[], {}, "", None]

    if isinstance(jsonData, list):
        jsonData = [new_elem for elem in jsonData
                    if (new_elem := remove_empty_elements(elem)) not in to_be_deleted]
   
    elif isinstance(jsonData,dict):
        jsonData = {key: new_value for key, value in jsonData.items()
                    if (new_value := remove_empty_elements(value)) not in to_be_deleted}

    return jsonData


class HelloFreshDumper:

    FMT_API_ENDPOINT = "https://gw.hellofresh.com/api/recipes/search?offset={}&order=-favorites&locale=en-GB&country=gb"
    PDF_NO_CONTENT_ADLER_32 = 29374378

    def __init__(self) -> None:
        # Variables
        self.session = None
        self.max_recipes = 0
        self.log = None
        self.collected_recipes = set()
        self.recipes_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'recipes')

        # Setup logging
        self._setup_logging()

        # Setup the session
        self._setup_session()

        # Create the recipes folder if it doesn't already exist
        if not os.path.exists(self.recipes_path):
            os.makedirs(self.recipes_path)
        else:
            self._setup_collected_recipes()

    def __del__(self) -> None:
        recipe_ids_path = os.path.join(self.recipes_path, "recipe_ids.json")
        with open(recipe_ids_path, 'w') as writer:
            json.dump(list(self.collected_recipes), writer)

    def _setup_logging(self):
        # Setup logging, with the format we want to use
        logging.basicConfig(format='%(name)s - %(levelname)s - %(message)s')
        # Create our logger
        self.log = logging.getLogger("HelloFreshDumper")
        # Set our logging level
        self.log.setLevel(logging.INFO)
        # Turn this down to just errors, otherwise we get spammed about
        # closing connections
        logging.getLogger("urllib3.connectionpool").setLevel(logging.ERROR)

    def _setup_session(self) -> None:
        self.log.debug("Retrieving authorization token")

        # Create the session object
        self.session = requests.Session()

        # Get a bearer token for the requests
        home = self.session.post("https://www.hellofresh.co.uk/gw/auth/token?client_id=senf&grant_type=client_credentials")
        
        # Get the access token
        auth = home.json()['access_token']

        # Set the authorization header
        self.session.headers["Authorization"] = "Bearer {}".format(auth)

        # Also get max_recipes from this web page
        self._set_max_recipes()

    def _setup_collected_recipes(self) -> None:
        recipe_ids_path = os.path.join(self.recipes_path, "recipe_ids.json")
        with open(recipe_ids_path, 'r') as reader:
            self.collected_recipes = set(json.load(reader))
    
    def _set_max_recipes(self) -> None:
        """Find the max number of recipes from the HelloFresh recipes homepage

        Args:
            page_text (string): text from the recipes hompage
        """
        url = "https://www.hellofresh.co.uk/gw/recipes/recipes/search?country=GB&locale=en-GB&skip=0&take=0"
        resp = self.session.get(url)
        self.max_recipes = resp.json()['total']
        self.log.info(f"{self.max_recipes} recipes available to collect in total")
        

    @staticmethod
    def _adjust_name(name: str) -> str:
        for x in ['\\', '/', ":", "<", ">", "?", "|"]:
            name = name.replace(x, '-')

        name = name.replace('"', "'")

        return name


    def _write_recipe(self, item) -> None:
        """Uses data from item to form a dictionary that contains just the
        data we need to write a useful recipe to disk and then writes that 
        dict to disk.
        - Cuts down on things we don't need.
        - Converts from IDs to actual ingredient/utensil names.

        Args:
            item (Dict): A recipe response dict/object from the request we made
                         from the API
        """
        # Useful fields
        required_fields = ['averageRating', 'cardLink', 'description', 'difficulty', 'favoritesCount',
                           'headline', 'steps', 'seoDescription', 'ratingsCount', 
                           'prepTime', 'link', 'imageLink']

        try:
            # Get all the easy data
            final = {k: v for k, v in item.items() if k in required_fields}

            # Get the category
            if item['category']:
                final['category'] = item['category']['name']
            
            # Convert utensil id's to actual names
            utils = {x['id']: x['name'] for x in item['utensils']}
            final['utensils'] = list(utils.values())
            for i, v in enumerate(item['steps']):
                final['steps'][i]['utensils'] = [utils[x] for x in v['utensils']]

            # Remove the keys we don't need for instructions
            for step in final['steps']:
                del step['instructionsHTML']
                del step['instructionsMarkdown']

                # Make a copy of instructions
                ingredient_ids = step['ingredients'][:]
                step['ingredients'] = [x['name'] for x in item['ingredients'] 
                                    if x['id'] in ingredient_ids]

            # Convert ingredient ids in yields to ingredient names
            # Get only the last/max yield item
            
            max_yield = item['yields'][-1]
            for ingredient_yields in max_yield['ingredients']:
                for ingredients_base in item['ingredients']:
                    if ingredient_yields['id'] == ingredients_base['id']:
                        ingredient_yields['name'] = ingredients_base['name']
                        del ingredient_yields['id']
                        break

            final['ingredients'] = max_yield['ingredients']
            final['serves'] = max_yield['yields']

        except Exception:   
            self.log.exception("Failed to parse the following object: {}".format(item.get('link', item)))
            # Exit out, we don't want to write anything to disk on failed recipes
            return

        # Clean up the dict
        final = remove_empty_elements(final)

        self.log.info(f"New recipe collected: {item['name']}")

        base_floc = "./recipes/{}".format(item['name'])
        # Write the file - Replace " with ' to make it a safe filename
        with open(base_floc + ".txt", 'w') as writer:
            # Prettify and write to disk
            json.dump(final, writer, indent=4, sort_keys=True)
        
        if item.get('cardLink'):
            try:
                content = self.session.get(item['cardLink']).content
                if zlib.adler32(content) != self.PDF_NO_CONTENT_ADLER_32:
                    with open(base_floc + ".pdf", 'wb') as writer:
                        writer.write(content)
            except Exception:
                self.log.exception("Failed to get card for: {}".format(item['name']))
                self.log.exception(final)

    def _recipe_request(self, offset):
        resp = self.session.get(self.FMT_API_ENDPOINT.format(offset))
        data = resp.json()

        items = []
        for item in data['items']:
            # Already collected, or recipes has no instructions
            item['name'] = self._adjust_name(item['name'])
            if item['id'] in self.collected_recipes or item['servingSize'] == 0:
                continue

            items.append(item)
            self.collected_recipes.add(item['id'])

        return items
    
    def _get_all_recipes(self):

        self.log.info("Retrieving all recipes")

        recipe_iter = range(1, self.max_recipes+1, 20)
        results = [item for sublist in 
                    tqdm.contrib.concurrent.thread_map(self._recipe_request, recipe_iter,
                                                       unit=' requests')
                    for item in sublist]

        return results

    def scrape(self):
        self.log.info("Scraping recipes")
        recipes = self._get_all_recipes()
        if not recipes:
            self.log.info("No new recipes to collect")
            return

        for x in recipes[:10]:
            self.log.info(f"Found: {x['name']}")
        if len(recipes) > 10:
            self.log.info(f"Found another {len(recipes)-10} recipes...")

        self.log.info("Parsing and writing recipes to disk")
        chunk_size=math.ceil(len(recipes)/1000)
        tqdm.contrib.concurrent.process_map(self._write_recipe, recipes, unit=' recipes', chunksize=chunk_size)
        self.log.info("Finished scraping recipes")

    def open_random_recipes(self, num_to_open: int) -> None:
        self.log.info(f"Opening up {num_to_open} random recipes")
        pdf_files = [x for x in os.listdir(self.recipes_path) if x.endswith(".pdf")]
        to_open = [os.path.join(self.recipes_path, secrets.choice(pdf_files)) for x in range(num_to_open)] 
        urls = [os.path.join(self.recipes_path, x) for x in to_open]
        for url in urls:
            # Most modern browsers should be able to open a pdf, you'd hope...
            webbrowser.open_new_tab(url)

def main():
    parser = argparse.ArgumentParser(description='Scrape recipies from HelloFresh')
    parser.add_argument('--no-scrape', action='store_true')
    args = parser.parse_args()

    scraper = HelloFreshDumper()
    if not args.no_scrape:
        scraper.scrape()
    scraper.open_random_recipes(3)

if __name__ == '__main__':
    main()