admin
f7810ba33d
- Add original recipe URL to Mealie description (appended after blank line) - Check for duplicate recipes on scrape (match orgURL or URL in description) - Show warning with link to existing recipe if duplicate found - User can still import anyway (warning only, not blocking) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
290 lines
9.9 KiB
Python
290 lines
9.9 KiB
Python
"""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, api_url: str = ""):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.api_url = api_url.rstrip("/") if api_url else self.base_url
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Accept": "application/json",
|
|
})
|
|
self._units_cache = None # {name_or_abbr: {id, name}}
|
|
self._foods_cache = {} # {lowercase_name: {id, name}}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_connection(self) -> dict:
|
|
"""Return Mealie app info or raise on failure."""
|
|
r = self.session.get(f"{self.api_url}/api/app/about", timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def find_duplicate(self, url: str, title: str = "") -> dict | None:
|
|
"""Check if a recipe with this original URL already exists.
|
|
|
|
Searches by URL in orgURL and description fields.
|
|
Returns {"slug": ..., "name": ..., "url": ...} or None.
|
|
"""
|
|
if not url:
|
|
return None
|
|
# Search by title to find candidates (Mealie search matches name)
|
|
search_term = title or url.split("/")[-1].replace("-", " ")
|
|
r = self.session.get(
|
|
f"{self.api_url}/api/recipes",
|
|
params={"search": search_term, "perPage": 50},
|
|
timeout=10,
|
|
)
|
|
if not r.ok:
|
|
return None
|
|
for item in r.json().get("items", []):
|
|
detail = self.session.get(
|
|
f"{self.api_url}/api/recipes/{item['slug']}", timeout=10
|
|
)
|
|
if not detail.ok:
|
|
continue
|
|
data = detail.json()
|
|
org = data.get("orgURL", "") or ""
|
|
desc = data.get("description", "") or ""
|
|
if org == url or url in desc:
|
|
return {
|
|
"slug": data["slug"],
|
|
"name": data.get("name", data["slug"]),
|
|
"url": f"{self.base_url}/g/home/r/{data['slug']}",
|
|
}
|
|
return None
|
|
|
|
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.api_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.api_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
|
|
|
|
# ------------------------------------------------------------------
|
|
# Unit / food resolution
|
|
# ------------------------------------------------------------------
|
|
|
|
def _load_units(self):
|
|
"""Fetch all units from Mealie, build lookup by name and abbreviation."""
|
|
if self._units_cache is not None:
|
|
return
|
|
self._units_cache = {}
|
|
r = self.session.get(f"{self.api_url}/api/units", params={"perPage": -1}, timeout=10)
|
|
r.raise_for_status()
|
|
for u in r.json().get("items", []):
|
|
entry = {"id": u["id"], "name": u["name"]}
|
|
self._units_cache[u["name"].lower()] = entry
|
|
abbr = u.get("abbreviation", "")
|
|
if abbr:
|
|
self._units_cache[abbr.lower()] = entry
|
|
|
|
def _ensure_unit(self, name: str):
|
|
"""Look up a unit by name/abbreviation. Create if missing.
|
|
Returns {"id": ..., "name": ...} or None for empty name."""
|
|
if not name:
|
|
return None
|
|
self._load_units()
|
|
key = name.lower().strip()
|
|
if key in self._units_cache:
|
|
return self._units_cache[key]
|
|
|
|
# Create new unit
|
|
r = self.session.post(
|
|
f"{self.api_url}/api/units",
|
|
json={"name": name, "abbreviation": name, "pluralName": name},
|
|
timeout=10,
|
|
)
|
|
if r.ok:
|
|
u = r.json()
|
|
entry = {"id": u["id"], "name": u["name"]}
|
|
self._units_cache[key] = entry
|
|
return entry
|
|
return None
|
|
|
|
def _ensure_food(self, name: str):
|
|
"""Look up a food by name. Create if missing.
|
|
Returns {"id": ..., "name": ...} or None for empty name."""
|
|
if not name:
|
|
return None
|
|
key = name.lower().strip()
|
|
if key in self._foods_cache:
|
|
return self._foods_cache[key]
|
|
|
|
# Search existing foods
|
|
r = self.session.get(
|
|
f"{self.api_url}/api/foods",
|
|
params={"search": name, "perPage": 50},
|
|
timeout=10,
|
|
)
|
|
if r.ok:
|
|
for f in r.json().get("items", []):
|
|
if f["name"].lower() == key:
|
|
entry = {"id": f["id"], "name": f["name"]}
|
|
self._foods_cache[key] = entry
|
|
return entry
|
|
|
|
# Create new food
|
|
r = self.session.post(
|
|
f"{self.api_url}/api/foods",
|
|
json={"name": name},
|
|
timeout=10,
|
|
)
|
|
if r.ok:
|
|
f = r.json()
|
|
entry = {"id": f["id"], "name": f["name"]}
|
|
self._foods_cache[key] = entry
|
|
return entry
|
|
return None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal
|
|
# ------------------------------------------------------------------
|
|
|
|
def _build_payload(self, recipe: dict) -> dict:
|
|
ingredients = []
|
|
pending_group = ""
|
|
for item in recipe.get("ingredients", []):
|
|
if isinstance(item, dict):
|
|
# Group header marker — apply title to the next real ingredient
|
|
if "group" in item and "food" not in item:
|
|
pending_group = item["group"]
|
|
continue
|
|
ing = self._build_ingredient(item)
|
|
else:
|
|
# Legacy: plain string
|
|
ing = {
|
|
"referenceId": str(uuid.uuid4()),
|
|
"note": str(item),
|
|
"isFood": False,
|
|
"disableAmount": True,
|
|
}
|
|
if pending_group:
|
|
ing["title"] = pending_group
|
|
pending_group = ""
|
|
ingredients.append(ing)
|
|
|
|
instructions = []
|
|
for text in recipe.get("instructions", []):
|
|
instructions.append({
|
|
"id": str(uuid.uuid4()),
|
|
"title": "",
|
|
"text": text,
|
|
"ingredientReferences": [],
|
|
})
|
|
|
|
description = recipe.get("description", "")
|
|
original_url = recipe.get("original_url", "")
|
|
if original_url and original_url not in description:
|
|
description = f"{description}\n\n{original_url}".strip()
|
|
|
|
return {
|
|
"name": recipe["title"],
|
|
"description": description,
|
|
"recipeIngredient": ingredients,
|
|
"recipeInstructions": instructions,
|
|
"orgURL": original_url,
|
|
"recipeYield": "",
|
|
}
|
|
|
|
def _build_ingredient(self, item: dict) -> dict:
|
|
"""Build a Mealie ingredient from a structured dict."""
|
|
qty_str = str(item.get("quantity", "")).strip()
|
|
unit_str = item.get("unit", "").strip()
|
|
food_str = item.get("food", "").strip()
|
|
extra = item.get("extra", "").strip()
|
|
|
|
has_structured = bool(qty_str or unit_str)
|
|
|
|
if has_structured and food_str:
|
|
unit_ref = self._ensure_unit(unit_str) if unit_str else None
|
|
food_ref = self._ensure_food(food_str)
|
|
|
|
qty = 0
|
|
try:
|
|
qty = float(qty_str.replace(",", "."))
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
return {
|
|
"referenceId": str(uuid.uuid4()),
|
|
"quantity": qty,
|
|
"unit": unit_ref,
|
|
"food": food_ref,
|
|
"note": extra,
|
|
"isFood": True,
|
|
"disableAmount": False,
|
|
}
|
|
else:
|
|
# No quantity/unit — put food+extra in note
|
|
note = food_str
|
|
if extra:
|
|
note = f"{note} ({extra})" if note else extra
|
|
return {
|
|
"referenceId": str(uuid.uuid4()),
|
|
"note": note,
|
|
"isFood": False,
|
|
"disableAmount": True,
|
|
}
|
|
|
|
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.api_url}/api/recipes/{slug}/image",
|
|
files=files,
|
|
data={"extension": ext},
|
|
timeout=30,
|
|
)
|
|
r.raise_for_status()
|