feat: Added initial version of the script
This commit is contained in:
295
recipe_bridge.py
Normal file
295
recipe_bridge.py
Normal file
@@ -0,0 +1,295 @@
|
||||
import datetime
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from argparse import ArgumentParser
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
class HttpResponder:
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
def json_request(self, url, method, headers=None, params=None, data=None):
|
||||
try:
|
||||
if method == "get":
|
||||
with requests.get(
|
||||
url=url, headers=headers, params=params
|
||||
) as r:
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
elif method == "post":
|
||||
with requests.post(
|
||||
url=url, headers=headers, params=params, json=data
|
||||
) as r:
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
elif method == "patch":
|
||||
with requests.patch(
|
||||
url=url, headers=headers, params=params, json=data
|
||||
) as r:
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
else:
|
||||
logging.error(f"Non supported/incorrect HTTP verb: {method}")
|
||||
exit(1)
|
||||
except requests.RequestException as e:
|
||||
logging.error(f"Request error: {e}")
|
||||
exit(1)
|
||||
except json.JSONDecodeError as e:
|
||||
logging.error(f"Error decoding response as JSON: {e}")
|
||||
exit(1)
|
||||
|
||||
|
||||
class HelloFresh(HttpResponder):
|
||||
def __init__(self, base_url, auth_token, country, language) -> None:
|
||||
self.base_url = base_url
|
||||
self.auth_token = auth_token
|
||||
self.recipes = set()
|
||||
self.headers = {
|
||||
"authorization": self.auth_token,
|
||||
"accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
self.params = {
|
||||
"country": country.upper(),
|
||||
"locale": f"{country.lower()}-{language.upper()}",
|
||||
}
|
||||
|
||||
def set_customer_id(self) -> None:
|
||||
customer_res = self.json_request(
|
||||
url=f"{self.base_url}/api/customers/me/subscriptions",
|
||||
method="get",
|
||||
headers=self.headers,
|
||||
params=self.params,
|
||||
)
|
||||
logging.debug(
|
||||
f'Adding customer ID {customer_res["items"][0]["id"]} to params'
|
||||
)
|
||||
self.params["subscription"] = customer_res["items"][0]["id"]
|
||||
|
||||
def set_current_week(self) -> None:
|
||||
today = datetime.date.today()
|
||||
logging.debug("Setting from HTTP parameter to current date")
|
||||
self.params["from"] = f"{today.year}-W{today.strftime('%V')}"
|
||||
|
||||
def add_monthly_recipes(self, deliveries) -> None:
|
||||
for weekly_delivery in deliveries["weeks"]:
|
||||
meals = weekly_delivery["meals"]
|
||||
for meal in meals:
|
||||
logging.debug(
|
||||
f'Getting HelloFresh recipe URL: {meal["websiteURL"]}'
|
||||
)
|
||||
self.recipes.add(meal["websiteURL"])
|
||||
|
||||
def get_past_deliveries(self, additional_deliveries) -> None:
|
||||
logging.debug("Getting last month deliveries")
|
||||
while True:
|
||||
most_recent_deliveries = self.json_request(
|
||||
url=f"{self.base_url}/my-deliveries/past-deliveries",
|
||||
method="get",
|
||||
headers=self.headers,
|
||||
params=self.params,
|
||||
)
|
||||
self.add_monthly_recipes(most_recent_deliveries)
|
||||
if additional_deliveries > 0:
|
||||
logging.debug("Getting more previous deliveries")
|
||||
try:
|
||||
self.params["from"] = most_recent_deliveries["nextWeek"]
|
||||
except KeyError:
|
||||
logging.error(
|
||||
f"Asked to retrieve {additional_deliveries} more months but no more deliveries found"
|
||||
)
|
||||
return
|
||||
additional_deliveries -= 1
|
||||
else:
|
||||
return
|
||||
|
||||
|
||||
class Mealie(HttpResponder):
|
||||
def __init__(self, base_url, auth_token) -> None:
|
||||
self.base_url = base_url
|
||||
self.auth_token = auth_token
|
||||
self.headers = {
|
||||
"Authorization": self.auth_token,
|
||||
"accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
self.tagged_recipes = []
|
||||
|
||||
def create_tag(self, tag) -> None:
|
||||
self.tag = self.json_request(
|
||||
url=f"{self.base_url}/api/organizers/tags",
|
||||
method="post",
|
||||
headers=self.headers,
|
||||
data={"name": tag},
|
||||
)
|
||||
logging.debug(f"Tag {tag} has been created")
|
||||
|
||||
def set_tag_id(self, tag) -> None:
|
||||
logging.debug(f"Getting {tag} tag infos")
|
||||
tag_id_res = self.json_request(
|
||||
url=f"{self.base_url}/api/organizers/tags",
|
||||
method="get",
|
||||
headers=self.headers,
|
||||
params={"search": tag},
|
||||
)
|
||||
if not tag_id_res["items"]:
|
||||
logging.info(f"Tag {tag} doesn't exist in Mealie, creating it")
|
||||
self.create_tag(tag)
|
||||
else:
|
||||
self.tag = tag_id_res["items"][0]
|
||||
|
||||
def get_tagged_recipes(self, tag) -> None:
|
||||
self.set_tag_id(tag)
|
||||
# Retrieve recipes count to infer paging
|
||||
tagged_recipes_nb_res = self.json_request(
|
||||
url=f"{self.base_url}/api/recipes",
|
||||
method="get",
|
||||
headers=self.headers,
|
||||
params={"tags": self.tag, "perPage": 0},
|
||||
)
|
||||
tagged_recipes_nb = tagged_recipes_nb_res["total"]
|
||||
|
||||
tagged_recipes_res = self.json_request(
|
||||
url=f"{self.base_url}/api/recipes",
|
||||
method="get",
|
||||
headers=self.headers,
|
||||
params={"tags": self.tag, "perPage": tagged_recipes_nb},
|
||||
)
|
||||
self.tagged_recipes = [
|
||||
recipe["orgURL"] for recipe in tagged_recipes_res["items"]
|
||||
]
|
||||
|
||||
def add_mealie_recipe(self, recipe_url) -> None:
|
||||
logging.info(f"Creating new recipe with url: {recipe_url}")
|
||||
new_recipe_slug = self.json_request(
|
||||
url=f"{self.base_url}/api/recipes/create/url",
|
||||
method="post",
|
||||
headers=self.headers,
|
||||
data={"url": recipe_url},
|
||||
)
|
||||
logging.debug("Getting newly created recipe ID")
|
||||
new_recipe_body = self.json_request(
|
||||
url=f"{self.base_url}/api/recipes/{new_recipe_slug}",
|
||||
method="get",
|
||||
headers=self.headers,
|
||||
)
|
||||
new_recipe_body["tags"].append(self.tag)
|
||||
logging.debug("Patching recipe to add custom tag to it")
|
||||
_ = self.json_request(
|
||||
url=f"{self.base_url}/api/recipes/{new_recipe_slug}",
|
||||
method="patch",
|
||||
headers=self.headers,
|
||||
data=new_recipe_body,
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
hellofresh_token = os.environ.get("hellofresh_token")
|
||||
if hellofresh_token == None:
|
||||
logging.error("Could not load required env var: HELLOFRESH_TOKEN")
|
||||
exit(1)
|
||||
mealie_token = os.environ.get("mealie_token")
|
||||
if mealie_token == None:
|
||||
logging.error("Could not load required env var: MEALIE_TOKEN")
|
||||
exit(1)
|
||||
|
||||
eligible_countries = [
|
||||
"at",
|
||||
"ch",
|
||||
"fr",
|
||||
"lu",
|
||||
"au",
|
||||
"de",
|
||||
"gb",
|
||||
"nl",
|
||||
"se",
|
||||
"be",
|
||||
"dk",
|
||||
"ie",
|
||||
"no",
|
||||
"us",
|
||||
"ca",
|
||||
"es",
|
||||
"it",
|
||||
"nz",
|
||||
]
|
||||
eligible_languages = ["de", "fr", "en", "nl", "sv", "da", "nb", "es", "it"]
|
||||
argParser = ArgumentParser()
|
||||
argParser.add_argument(
|
||||
"--country",
|
||||
"-c",
|
||||
help="Country linked to your HelloFresh account",
|
||||
choices=eligible_countries,
|
||||
required=True,
|
||||
)
|
||||
argParser.add_argument(
|
||||
"--language",
|
||||
"-l",
|
||||
help="Locale linked to your HelloFresh account",
|
||||
choices=eligible_languages,
|
||||
required=True,
|
||||
)
|
||||
argParser.add_argument(
|
||||
"--mealie-tag",
|
||||
"-t",
|
||||
help="Mealie tag to group HelloFresh recipes (default: HelloFresh)",
|
||||
default="HelloFresh",
|
||||
)
|
||||
argParser.add_argument(
|
||||
"--additional-deliveries",
|
||||
"-a",
|
||||
help="Number of additional months to retrieve from HelloFresh",
|
||||
type=int,
|
||||
default=0,
|
||||
)
|
||||
argParser.add_argument(
|
||||
"--debug", help="Enable debug logs", action="store_true"
|
||||
)
|
||||
argParser.add_argument(
|
||||
"--dry-run",
|
||||
"-d",
|
||||
help="Just fetch and count recipes from HelloFresh that would be added to Mealie",
|
||||
action="store_true",
|
||||
)
|
||||
args = argParser.parse_args()
|
||||
if args.debug:
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
else:
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
hellofresh_api_url = f"https://www.hellofresh.{args.country.lower()}/gw"
|
||||
hellofresh_client = HelloFresh(
|
||||
hellofresh_api_url, hellofresh_token, args.country, args.language
|
||||
)
|
||||
hellofresh_client.set_customer_id()
|
||||
hellofresh_client.set_current_week()
|
||||
hellofresh_client.get_past_deliveries(args.additional_deliveries)
|
||||
logging.info(
|
||||
f"Scrapped {len(hellofresh_client.recipes)} HelloFresh recipes"
|
||||
)
|
||||
|
||||
mealie_api_url = "https://food.syyrell.com"
|
||||
mealie_client = Mealie(mealie_api_url, mealie_token)
|
||||
mealie_client.get_tagged_recipes(args.mealie_tag)
|
||||
if args.dry_run:
|
||||
new_recipes = hellofresh_client.recipes - set(
|
||||
mealie_client.tagged_recipes
|
||||
)
|
||||
if len(new_recipes) > 0:
|
||||
logging.info(
|
||||
f"Would have added {len(new_recipes)} recipes to Mealie:"
|
||||
)
|
||||
{logging.info(recipe) for recipe in new_recipes}
|
||||
else:
|
||||
logging.info("All fetched recipes already exist in Mealie!")
|
||||
exit(0)
|
||||
for new_recipe in hellofresh_client.recipes:
|
||||
if new_recipe not in mealie_client.tagged_recipes:
|
||||
mealie_client.add_mealie_recipe(new_recipe)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user