mealie-mcp/mealie_mcp/server.py

"""Small, confirmation-friendly MCP facade over the Mealie API."""
from __future__ import annotations

import json
import os
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path
from typing import Any

import httpx
from mcp.server.mcpserver import MCPServer

from .normalize import (
    clean_title,
    has_extraction_failure,
    ingredient_display_lines,
    looks_mangled,
    normalize_for_parser,
)
from .verify import verify_recipe as run_verification

BASE_URL = os.getenv("MEALIE_BASE_URL", "https://recept.famfallman.com").rstrip("/")
TOKEN = os.getenv("MEALIE_API_TOKEN")
UA = os.getenv("MEALIE_USER_AGENT", "Mealie-MCP/0.1")
HA_BASE_URL = (os.getenv("HA_BASE_URL") or "").rstrip("/")
HA_TOKEN = os.getenv("HA_TOKEN")
SHOPPING_LIST_ENTITY = os.getenv("HA_SHOPPING_LIST_ENTITY", "todo.ourgroceries_shoppinglista")

RESOURCE_DIR = Path(__file__).parent / "resources"

mcp = MCPServer("mealie", version="0.1.0")


class MealieAuthError(RuntimeError):
    """The Mealie token is missing, expired, or wrong (HTTP 401)."""


class MealieBlockedError(RuntimeError):
    """The edge/WAF rejected the client fingerprint (HTTP 403, Cloudflare 1010)."""


def _client() -> httpx.Client:
    if not TOKEN:
        raise RuntimeError("MEALIE_API_TOKEN is not set")
    return httpx.Client(
        base_url=BASE_URL,
        headers={"Authorization": f"Bearer {TOKEN}", "User-Agent": UA},
        timeout=30,
    )


def _raise_for_status(response: httpx.Response) -> None:
    """Map Mealie/Cloudflare failures to distinguishable errors.

    A 403 with Cloudflare's ``error code: 1010`` is a blocked client fingerprint,
    not a bad token — conflating the two sends debugging in the wrong direction.
    """
    if response.status_code == 401:
        raise MealieAuthError(
            "401 from Mealie: the API token is missing, expired, or wrong. "
            "Create a new one at /user/profile/api-tokens."
        )
    if response.status_code == 403 and "1010" in response.text:
        raise MealieBlockedError(
            f"403 from the edge (Cloudflare error 1010) for User-Agent {UA!r}. "
            "This is a WAF rule on the client fingerprint, not a Mealie auth problem. "
            "Set MEALIE_USER_AGENT to a non-default value."
        )
    response.raise_for_status()


def _request(method: str, path: str, **kwargs: Any) -> Any:
    with _client() as client:
        response = client.request(method, path, **kwargs)
        _raise_for_status(response)
        if not response.content:
            return None
        return response.json()


# --- Read -------------------------------------------------------------------


@mcp.tool()
def check_auth() -> dict[str, Any]:
    """Verify the Mealie token and distinguish auth failures from WAF blocks."""
    try:
        user = _request("GET", "/api/users/self")
    except MealieAuthError as exc:
        return {"ok": False, "reason": "auth", "detail": str(exc)}
    except MealieBlockedError as exc:
        return {"ok": False, "reason": "waf", "detail": str(exc)}
    return {
        "ok": True,
        "base_url": BASE_URL,
        "user": user.get("username"),
        "user_agent": UA,
    }


@mcp.tool()
def search_recipes(query: str, limit: int = 10) -> list[dict[str, Any]]:
    """Search Mealie recipes by text."""
    data = _request("GET", "/api/recipes", params={"search": query, "perPage": limit})
    return data.get("items", data if isinstance(data, list) else [])


@mcp.tool()
def suggest_recipes(foods: list[str], limit: int = 10, max_missing_foods: int = 0) -> Any:
    """Find recipes matching ingredients on hand."""
    return _request("GET", "/api/recipes/suggestions", params={
        "foods": ",".join(foods), "limit": limit,
        "maxMissingFoods": max_missing_foods, "includeFoodsOnHand": "true",
    })


@mcp.tool()
def get_recipe(slug_or_id: str) -> dict[str, Any]:
    """Read complete recipe details."""
    return _request("GET", f"/api/recipes/{slug_or_id}")


@mcp.tool()
def list_organizers() -> dict[str, list[str]]:
    """List existing categories and tags so imports reuse the library's vocabulary.

    Read this before tagging anything: the library has settled naming (`Källa: Köket`,
    `📅 Vardag`, `💨 Airfryer`) and inventing parallel names fragments retrieval.
    """
    def names(path: str) -> list[str]:
        data = _request("GET", path, params={"perPage": 200})
        items = data.get("items", data if isinstance(data, list) else [])
        return sorted(item.get("name", "") for item in items)

    return {
        "categories": names("/api/organizers/categories"),
        "tags": names("/api/organizers/tags"),
    }


@mcp.tool()
def find_by_source_url(url: str) -> list[dict[str, Any]]:
    """Look for an existing recipe with this source URL, to avoid duplicate imports.

    ``orgURL`` is the reliable duplicate key; imported titles and slugs legitimately
    differ from the URL text, so matching on those produces false negatives.
    """
    data = _request("GET", "/api/recipes", params={"perPage": 500})
    items = data.get("items", data if isinstance(data, list) else [])
    return [item for item in items if (item.get("orgURL") or "").rstrip("/") == url.rstrip("/")]


@mcp.tool()
def scale_ingredients(slug_or_id: str, servings: int) -> dict[str, Any]:
    """Return readable ingredient lines scaled to a requested serving count."""
    recipe = get_recipe(slug_or_id)
    original = recipe.get("recipeYield") or recipe.get("recipeServings")
    if not original:
        raise ValueError("Recipe has no serving/yield metadata; scaling is not exact")
    try:
        base = Decimal(str(original).split()[0])
        factor = Decimal(servings) / base
    except Exception as exc:
        raise ValueError("Recipe serving metadata is not numeric") from exc
    lines = []
    for item in recipe.get("recipeIngredient", []):
        text = item.get("display") or item.get("note") or item.get("food", {}).get("name", "")
        quantity = item.get("quantity")
        if quantity is not None:
            scaled = (Decimal(str(quantity)) * factor).quantize(Decimal("0.1"), rounding=ROUND_HALF_UP)
            text = text.replace(str(quantity), format(scaled, "f"), 1)
        lines.append(text)
    return {"recipe": recipe.get("name"), "servings": servings, "ingredients": lines}


# --- Import -----------------------------------------------------------------


def _import_report(recipe: dict[str, Any]) -> dict[str, Any]:
    """Flag extraction problems the caller must fix before claiming success."""
    title = recipe.get("name") or ""
    cleaned = clean_title(title)
    return {
        "slug": recipe.get("slug"),
        "title": title,
        "suggested_title": cleaned if cleaned != title else None,
        "extraction_failed": has_extraction_failure(recipe),
        "has_image_field": bool(recipe.get("image")),
        "ingredient_count": len(recipe.get("recipeIngredient") or []),
        "instruction_count": len(recipe.get("recipeInstructions") or []),
    }


@mcp.tool()
def import_recipe_url(url: str, check_duplicates: bool = True) -> dict[str, Any]:
    """Import a recipe page through Mealie's URL scraper.

    Returns the created recipe plus an extraction report. A non-empty response is
    not success: check ``extraction_failed`` and ``suggested_title`` before
    reporting, and run the Swedish normalization pass afterwards.
    """
    if check_duplicates:
        existing = find_by_source_url(url)
        if existing:
            return {
                "imported": False,
                "reason": "duplicate",
                "existing": [{"slug": r.get("slug"), "name": r.get("name")} for r in existing],
            }
    created = _request("POST", "/api/recipes/create/url", json={"url": url})
    recipe = get_recipe(created) if isinstance(created, str) else created
    return {"imported": True, "recipe": recipe, "report": _import_report(recipe)}


@mcp.tool()
def import_recipe_text(text: str, source_url: str | None = None) -> dict[str, Any]:
    """Import a normalized recipe text block (OCR, transcript, or hand-built Swedish).

    Use this when the source is not a clean single-recipe page — an editorial
    roundup, a YouTube description, or a social caption. Build the text in Swedish
    with ``Titel`` / ``Portioner`` / ``Ingredienser`` / ``Gör så här`` sections.
    """
    payload: dict[str, Any] = {"data": text}
    if source_url:
        payload["url"] = source_url
    created = _request("POST", "/api/recipes/create/html-or-json", json=payload)
    recipe = get_recipe(created) if isinstance(created, str) else created
    return {"imported": True, "recipe": recipe, "report": _import_report(recipe)}


@mcp.tool()
def import_recipe_image(image_path: str) -> dict[str, Any]:
    """Import a recipe from a photo or screenshot via Mealie's image parser."""
    path = Path(image_path)
    if not path.is_file():
        raise ValueError(f"No such image file: {image_path}")
    with _client() as client, path.open("rb") as handle:
        response = client.post("/api/recipes/create/image", files={"images": (path.name, handle)})
        _raise_for_status(response)
        created = response.json()
    recipe = get_recipe(created) if isinstance(created, str) else created
    return {"imported": True, "recipe": recipe, "report": _import_report(recipe)}


# --- Write ------------------------------------------------------------------


@mcp.tool()
def patch_recipe(slug_or_id: str, patch: dict[str, Any]) -> dict[str, Any]:
    """Apply an explicit Mealie recipe patch, then read the recipe back.

    Use this for tags and categories too — the bulk-action endpoints return 500 on
    this instance. Patch ``recipeCategory`` and ``tags`` directly.
    """
    _request("PATCH", f"/api/recipes/{slug_or_id}", json=patch)
    return get_recipe(slug_or_id)


@mcp.tool()
def parse_ingredients(slug_or_id: str) -> dict[str, Any]:
    """Add structured food/unit/quantity data while keeping the Swedish display text.

    The parser is a structured-data helper only. Its ``display`` output mangles
    Swedish rows (``4,7 dl basmatiris`` came back as ``47⁄10 liter ...``), so the
    canonical human line is preserved and restored after parsing. Rows the parser
    still mangles keep readable text with structure cleared — legibility wins.
    """
    recipe = get_recipe(slug_or_id)
    canonical = ingredient_display_lines(recipe)
    if not canonical:
        return {"parsed": False, "reason": "Recipe has no ingredient lines"}

    parsed = _request(
        "POST",
        "/api/parser/ingredients",
        json={"ingredients": [normalize_for_parser(line) for line in canonical], "parser": "nlp"},
    )

    ingredients = list(recipe.get("recipeIngredient") or [])
    degraded: list[str] = []
    for index, result in enumerate(parsed or []):
        if index >= len(ingredients):
            break
        item = dict(ingredients[index])
        confidence = (result or {}).get("ingredient") or {}
        item["quantity"] = confidence.get("quantity")
        item["unit"] = confidence.get("unit")
        item["food"] = confidence.get("food")
        # The canonical Swedish line always wins over parser-generated display text.
        item["display"] = canonical[index]
        item["note"] = canonical[index]
        if looks_mangled(canonical[index]):
            item["quantity"] = None
            item["unit"] = None
            item["food"] = None
            degraded.append(canonical[index])
        ingredients[index] = item

    try:
        _request("PATCH", f"/api/recipes/{slug_or_id}", json={"recipeIngredient": ingredients})
    except httpx.HTTPStatusError as exc:
        # A full recipeIngredient patch with nested parser objects can 500 here.
        # Preserve the readable import rather than corrupting the recipe.
        return {
            "parsed": False,
            "reason": f"Structured patch rejected by Mealie ({exc.response.status_code}); "
                      "readable Swedish lines were left intact",
            "canonical_lines": canonical,
        }

    return {
        "parsed": True,
        "recipe": get_recipe(slug_or_id),
        "readability_fallback_rows": degraded,
    }


@mcp.tool()
def set_cover_image(
    slug_or_id: str,
    source_url: str | None = None,
    image_path: str | None = None,
) -> dict[str, Any]:
    """Set a recipe cover image by scraping a URL or uploading a local file.

    Local uploads must send the multipart ``extension`` field or the upload fails
    validation silently on this instance. When a scrape does not produce a visible
    cover (observed on Allrecipes), download the hero image and upload it instead.
    """
    if not source_url and not image_path:
        raise ValueError("Provide either source_url or image_path")

    if image_path:
        path = Path(image_path)
        if not path.is_file():
            raise ValueError(f"No such image file: {image_path}")
        extension = path.suffix.lstrip(".").lower() or "jpg"
        with _client() as client, path.open("rb") as handle:
            response = client.put(
                f"/api/recipes/{slug_or_id}/image",
                files={"image": (path.name, handle)},
                data={"extension": extension},
            )
            _raise_for_status(response)
        method = f"upload ({extension})"
    else:
        _request("POST", f"/api/recipes/{slug_or_id}/image", json={"url": source_url})
        method = "scrape"

    recipe = get_recipe(slug_or_id)
    return {"method": method, "image": recipe.get("image"), "slug": recipe.get("slug")}


# --- Verify -----------------------------------------------------------------


@mcp.tool()
def verify_recipe(
    slug_or_id: str,
    image_verified: bool = False,
    taxonomy_skipped_reason: str | None = None,
) -> dict[str, Any]:
    """Run the finished-import checklist and return pass/fail per check.

    Set ``image_verified`` only if you actually ran an image step — a non-empty
    ``image`` field does not prove the UI shows a cover.
    """
    recipe = get_recipe(slug_or_id)
    return run_verification(
        recipe,
        image_verified=image_verified,
        taxonomy_skipped_reason=taxonomy_skipped_reason,
    )


# --- Home Assistant ---------------------------------------------------------


@mcp.tool()
def shopping_list_add(items: list[str]) -> dict[str, Any]:
    """Add ingredient lines to the Home Assistant shopping list.

    Each line must carry its quantity — the todo entity has no structured quantity
    field, so `400 g kycklinglårfilé`, not `kycklinglårfilé`. Duplicates are never
    silently removed; consolidation is a separate, explicit operation.
    """
    if not HA_BASE_URL or not HA_TOKEN:
        raise RuntimeError("HA_BASE_URL and HA_TOKEN must be set to write the shopping list")
    added: list[str] = []
    with httpx.Client(
        base_url=HA_BASE_URL,
        headers={"Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json"},
        timeout=30,
    ) as client:
        for item in items:
            response = client.post(
                "/api/services/todo/add_item",
                json={"entity_id": SHOPPING_LIST_ENTITY, "item": item},
            )
            response.raise_for_status()
            added.append(item)
    return {"entity_id": SHOPPING_LIST_ENTITY, "added": added, "count": len(added)}


# --- Resources and prompts --------------------------------------------------


def _resource(filename: str) -> str:
    return (RESOURCE_DIR / filename).read_text(encoding="utf-8")


@mcp.resource("mealie://reference/family-taste-profile")
def family_taste_profile() -> str:
    """Fredrik's household taste profile, for judging whether a recipe is a good fit."""
    return _resource("family-taste-profile.md")


@mcp.resource("mealie://reference/import-pitfalls")
def import_pitfalls() -> str:
    """Verified post-import failure modes: image, parser, bulk-action, and decimal issues."""
    return _resource("import-pitfalls.md")


@mcp.resource("mealie://reference/koket-attribution")
def koket_attribution() -> str:
    """How to preserve Köket author/programme attribution after import."""
    return _resource("koket-attribution.md")


@mcp.resource("mealie://reference/cloudflare-hostname")
def cloudflare_hostname() -> str:
    """Why the external hostname returns 403/1010 and how to avoid it."""
    return _resource("cloudflare-hostname.md")


@mcp.prompt()
def import_pipeline(source: str) -> str:
    """The full import order, so no step gets skipped."""
    return f"""Import this source into Mealie: {source}

Follow this order exactly. Skipping a step is how imports end up half-finished.

1. `check_auth` if you have not already this session.
2. `find_by_source_url` — stop and report if it already exists.
3. Import with `import_recipe_url` (single recipe page) or `import_recipe_text`
   (editorial roundup, transcript, caption). Read the returned report: if
   `extraction_failed` is true, rebuild the recipe manually instead of patching.
4. `list_organizers` to see the existing category/tag vocabulary.
5. `patch_recipe` with Swedish title, description, instructions, and ingredient
   lines, plus `orgURL`, `extras.author`, `extras.source`, `tags`, and
   `recipeCategory`. Apply `suggested_title` if the report offered one.
6. `set_cover_image` explicitly — do not trust a non-empty `image` field.
7. `parse_ingredients` to add structure while keeping the Swedish display text.
8. `verify_recipe` with `image_verified=True`. Do not report success until
   `complete` is true, or state exactly which checks failed and why.

All stored values must be Swedish: title, description, ingredient names,
instructions, and units (`msk`, `tsk`, `dl`, `krm`, `st`, `klyfta`).
"""


def main() -> None:
    mcp.run()