"""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 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 = [] for item in recipe.get("ingredients", []): if isinstance(item, dict): ingredients.append(self._build_ingredient(item)) else: # Legacy: plain string ingredients.append({ "referenceId": str(uuid.uuid4()), "note": str(item), "isFood": False, "disableAmount": True, }) instructions = [] for text in recipe.get("instructions", []): instructions.append({ "id": str(uuid.uuid4()), "title": "", "text": text, "ingredientReferences": [], }) return { "name": recipe["title"], "description": recipe.get("description", ""), "recipeIngredient": ingredients, "recipeInstructions": instructions, "orgURL": recipe.get("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, timeout=30, ) r.raise_for_status()