feat: initial recipe-importer application
Python/Flask web app that scrapes Hungarian recipe sites (mindmegette.hu) and imports them into Mealie via its REST API. Includes dark-themed web UI with editable preview, Dockerfile, build script, and docker-compose. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
+114
@@ -0,0 +1,114 @@
|
||||
"""Mealie API client — creates recipes and uploads images."""
|
||||
|
||||
import io
|
||||
import uuid
|
||||
import requests
|
||||
|
||||
|
||||
class MealieClient:
|
||||
"""Thin wrapper around the Mealie REST API."""
|
||||
|
||||
def __init__(self, base_url: str, api_key: str):
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.session = requests.Session()
|
||||
self.session.headers.update({
|
||||
"Authorization": f"Bearer {api_key}",
|
||||
"Accept": "application/json",
|
||||
})
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_connection(self) -> dict:
|
||||
"""Return Mealie app info or raise on failure."""
|
||||
r = self.session.get(f"{self.base_url}/api/app/about", timeout=10)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
def create_recipe(self, recipe: dict) -> str:
|
||||
"""Create a recipe in Mealie from a scraper result dict.
|
||||
|
||||
*recipe* keys: title, description, image_url, ingredients, instructions, original_url.
|
||||
Returns the recipe slug.
|
||||
"""
|
||||
# Step 1: create stub
|
||||
r = self.session.post(
|
||||
f"{self.base_url}/api/recipes",
|
||||
json={"name": recipe["title"]},
|
||||
timeout=15,
|
||||
)
|
||||
r.raise_for_status()
|
||||
slug = r.json() # Mealie returns the slug as a plain string
|
||||
|
||||
# Step 2: build full payload and PATCH
|
||||
payload = self._build_payload(recipe)
|
||||
r = self.session.patch(
|
||||
f"{self.base_url}/api/recipes/{slug}",
|
||||
json=payload,
|
||||
timeout=15,
|
||||
)
|
||||
r.raise_for_status()
|
||||
|
||||
# Step 3: upload image if available
|
||||
image_url = recipe.get("image_url")
|
||||
if image_url:
|
||||
try:
|
||||
self._upload_image(slug, image_url)
|
||||
except Exception:
|
||||
pass # non-fatal — recipe is still created
|
||||
|
||||
return slug
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _build_payload(self, recipe: dict) -> dict:
|
||||
ingredients = []
|
||||
for line in recipe.get("ingredients", []):
|
||||
ingredients.append({
|
||||
"note": line,
|
||||
"isFood": False,
|
||||
"disableAmount": True,
|
||||
})
|
||||
|
||||
instructions = []
|
||||
for text in recipe.get("instructions", []):
|
||||
instructions.append({
|
||||
"id": uuid.uuid4().hex,
|
||||
"text": text,
|
||||
})
|
||||
|
||||
return {
|
||||
"name": recipe["title"],
|
||||
"description": recipe.get("description", ""),
|
||||
"recipeIngredient": ingredients,
|
||||
"recipeInstructions": instructions,
|
||||
"orgURL": recipe.get("original_url", ""),
|
||||
"recipeYield": "",
|
||||
}
|
||||
|
||||
def _upload_image(self, slug: str, image_url: str):
|
||||
"""Download image from *image_url* and upload it to the recipe."""
|
||||
img_resp = requests.get(image_url, timeout=30, headers={
|
||||
"User-Agent": "RecipeImporter/1.0",
|
||||
})
|
||||
img_resp.raise_for_status()
|
||||
|
||||
content_type = img_resp.headers.get("Content-Type", "image/jpeg")
|
||||
ext = "jpg"
|
||||
if "png" in content_type:
|
||||
ext = "png"
|
||||
elif "webp" in content_type:
|
||||
ext = "webp"
|
||||
|
||||
files = {
|
||||
"image": (f"recipe.{ext}", io.BytesIO(img_resp.content), content_type),
|
||||
}
|
||||
r = self.session.put(
|
||||
f"{self.base_url}/api/recipes/{slug}/image",
|
||||
files=files,
|
||||
timeout=30,
|
||||
)
|
||||
r.raise_for_status()
|
||||
Reference in New Issue
Block a user