mealie-mcp/mealie_mcp/server.py

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

import html
import json
import os
import re
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,
    food_matches_line,
    has_extraction_failure,
    ingredient_display_lines,
    line_remainder,
    looks_mangled,
    match_unit,
    normalize_for_parser,
    same_line,
    scale_line,
    unit_token,
)
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")

WRITE_TOOL_NAMES = frozenset(
    {
        "import_recipe_url",
        "import_recipe_text",
        "import_recipe_image",
        "patch_recipe",
        "delete_recipe",
        "parse_ingredients",
        "set_cover_image",
        "shopping_list_add",
    }
)


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 resolve_foods(names: list[str]) -> dict[str, Any]:
    """Map food names to the Mealie food records they refer to.

    ``/api/recipes/suggestions`` takes food UUIDs, not names, so free-text has to
    be resolved first. Returns the best match per name plus the alternatives, so
    the caller can tell that `kyckling` resolved to `kycklingbröst`.
    """
    resolved: dict[str, Any] = {}
    unresolved: list[str] = []
    for name in names:
        data = _request("GET", "/api/foods", params={"search": name, "perPage": 5})
        items = data.get("items", data if isinstance(data, list) else [])
        if not items:
            unresolved.append(name)
            continue
        exact = next((i for i in items if (i.get("name") or "").lower() == name.lower()), None)
        best = exact or items[0]
        resolved[name] = {
            "id": best.get("id"),
            "name": best.get("name"),
            "exact": exact is not None,
            "alternatives": [i.get("name") for i in items if i.get("id") != best.get("id")],
        }
    return {"resolved": resolved, "unresolved": unresolved}


@mcp.tool()
def suggest_recipes(foods: list[str], limit: int = 10, max_missing_foods: int = 5) -> dict[str, Any]:
    """Find recipes matching ingredients on hand, given plain food names.

    The endpoint expects food UUIDs as repeated query parameters; passing names or
    a comma-joined string returns 422. Names are resolved first, and anything that
    could not be resolved is reported rather than silently dropped.
    """
    lookup = resolve_foods(foods)
    ids = [entry["id"] for entry in lookup["resolved"].values()]
    if not ids:
        return {"items": [], "matched_foods": {}, "unresolved_foods": lookup["unresolved"]}
    data = _request("GET", "/api/recipes/suggestions", params=[
        *[("foods", food_id) for food_id in ids],
        ("limit", limit),
        ("maxMissingFoods", max_missing_foods),
        ("includeFoodsOnHand", "true"),
    ])
    items = data.get("items", data if isinstance(data, list) else [])
    return {
        "items": items,
        "matched_foods": {name: e["name"] for name, e in lookup["resolved"].items()},
        "unresolved_foods": lookup["unresolved"],
    }


@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.

    Amounts are read from the display text, not the structured ``quantity`` field:
    unparsed recipes store ``quantity: 0.0`` while the real amount lives in the
    line. Lines with no leading amount (`salt`, `smör (att steka i)`) are returned
    untouched and listed in ``unscaled`` — never silently presented as scaled.
    """
    recipe = get_recipe(slug_or_id)
    original = recipe.get("recipeServings") or recipe.get("recipeYieldQuantity")
    if not original:
        # recipeYield is free text ("4 portioner"); take a leading number if present.
        match = re.match(r"\s*(\d+(?:[.,]\d+)?)", str(recipe.get("recipeYield") or ""))
        original = match.group(1).replace(",", ".") if match else None
    if not original or Decimal(str(original)) == 0:
        raise ValueError(
            "Recipe has no serving metadata (recipeServings, recipeYieldQuantity, "
            "or a numeric recipeYield); scaling would be a guess"
        )

    factor = Decimal(servings) / Decimal(str(original))
    scaled_lines: list[str] = []
    unscaled: list[str] = []
    for line in ingredient_display_lines(recipe):
        text, was_scaled = scale_line(line, factor)
        scaled_lines.append(text)
        if not was_scaled and line.strip():
            unscaled.append(line)

    return {
        "recipe": recipe.get("name"),
        "original_servings": float(Decimal(str(original))),
        "servings": servings,
        "factor": float(round(factor, 3)),
        "ingredients": scaled_lines,
        "unscaled": unscaled,
    }


# --- 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.
    """
    required = ("Titel", "Ingredienser", "Gör så här")
    missing = [
        heading
        for heading in required
        if not re.search(rf"(?im)^\s*{re.escape(heading)}\s*:", text)
    ]
    if missing:
        raise ValueError(
            "Text import requires explicit Swedish section headings; missing: "
            + ", ".join(missing)
        )
    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. The stable
    recipe id is captured before the PATCH because changing ``name`` also changes
    Mealie's slug, making the old slug invalid for read-back.
    """
    before = get_recipe(slug_or_id)
    stable_id = before.get("id") or slug_or_id
    _request("PATCH", f"/api/recipes/{slug_or_id}", json=patch)
    return get_recipe(stable_id)


@mcp.tool()
def delete_recipe(slug_or_id: str, confirm_slug: str) -> dict[str, Any]:
    """Delete exactly one recipe, with explicit slug confirmation and read-back proof.

    This is primarily the rollback path for failed imports and marked test recipes.
    The caller must first read the recipe and pass its exact stored slug as
    ``confirm_slug``. Names and ids are not accepted as confirmation.
    """
    recipe = get_recipe(slug_or_id)
    stored_slug = recipe.get("slug") or ""
    if confirm_slug != stored_slug:
        raise ValueError(
            f"Deletion confirmation does not match stored slug {stored_slug!r}"
        )

    with _client() as client:
        response = client.delete(f"/api/recipes/{stored_slug}")
        _raise_for_status(response)
        probe = client.get(f"/api/recipes/{stored_slug}")
    if probe.status_code != 404:
        raise RuntimeError(
            f"Delete returned {response.status_code}, but read-back returned "
            f"{probe.status_code} instead of 404"
        )
    return {
        "deleted": True,
        "slug": stored_slug,
        "name": recipe.get("name"),
        "delete_status": response.status_code,
        "readback_status": probe.status_code,
    }


def _unit_records() -> list[dict[str, Any]]:
    """The instance's own unit table — the authority for what `dl` and `msk` mean."""
    data = _request("GET", "/api/units", params={"perPage": 200})
    return data.get("items", data if isinstance(data, list) else [])


def _vet_structure(line: str, ingredient: dict[str, Any], units: list[dict[str, Any]]) -> tuple[
    Any, dict[str, Any] | None, dict[str, Any] | None, list[str]
]:
    """Keep only parser output the Swedish line actually supports.

    Verified against the live instance: the parser resolves `3 dl` to the `liter`
    record, swaps `kycklinglårfilé` for the existing food `kycklingfilé`, and
    invents foods with ``id: null`` — the last of which makes the recipe PATCH
    fail with a 500. Unsupported values are dropped, never guessed at, and every
    drop is reported so the caller can see what was not structured.
    """
    quantity = ingredient.get("quantity")
    unit = ingredient.get("unit")
    food = ingredient.get("food")
    notes: list[str] = []

    if looks_mangled(line):
        return None, None, None, ["stored line is already parser-damaged; left unstructured"]

    if unit:
        written = unit_token(line)
        confirmed = match_unit(written, units)
        if confirmed and confirmed.get("id") != unit.get("id"):
            notes.append(
                f"unit {unit.get('name')!r} corrected to {confirmed.get('name')!r} "
                f"from the written token {written!r}"
            )
            unit = confirmed
        elif not confirmed:
            notes.append(
                f"unit {unit.get('name')!r} is not what the line writes "
                f"({written!r}); dropped"
            )
            unit = None

    if food and not food.get("id"):
        # A food with no id makes Mealie 500 on the recipe patch (ValueError).
        notes.append(f"food {food.get('name')!r} does not exist in Mealie; dropped")
        food = None
    elif food and not food_matches_line(food.get("name") or "", line):
        notes.append(f"food {food.get('name')!r} is not the ingredient this line names; dropped")
        food = None

    return quantity, unit, food, notes


def _unstructured_item(base: dict[str, Any], line: str) -> dict[str, Any]:
    """A row carrying only the human line — Mealie renders ``note`` verbatim."""
    item = dict(base)
    item["quantity"] = 0
    item["unit"] = None
    item["food"] = None
    item["note"] = line
    return item


def _structured_item(
    base: dict[str, Any], line: str, quantity: Any, unit: Any, food: Any
) -> dict[str, Any]:
    """A row holding structure, with ``note`` reduced to the leftover text."""
    item = dict(base)
    item["quantity"] = quantity
    item["unit"] = unit
    item["food"] = food
    item["note"] = line_remainder(line, unit_token(line) if unit else None,
                                  (food or {}).get("name") if food else None)
    return item


@mcp.tool()
def parse_ingredients(slug_or_id: str) -> dict[str, Any]:
    """Add structured food/unit/quantity data without changing any visible Swedish line.

    ``display`` is computed by Mealie from ``quantity unit food note`` and cannot be
    set through the API — writing it is silently ignored. Structuring a row
    therefore rewrites how it reads, and Mealie's rendering is not always the
    Swedish the recipe was written in (``800 g`` renders as ``800 gram``, ``0,25``
    as ``¹/₄``, and plural foods render singular).

    So structure is applied, read back, and kept only for rows Mealie renders
    identically to the original line. Any row it would have reworded is restored
    unstructured, and reported in ``left_unstructured`` with the text Mealie wanted
    to show. The visible line is never changed by this tool.
    """
    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"},
    )
    units = _unit_records()

    base = list(recipe.get("recipeIngredient") or [])
    candidates = [_unstructured_item(item, line) for item, line in zip(base, canonical)]
    corrections: list[dict[str, Any]] = []
    for index, result in enumerate(parsed or []):
        if index >= len(base):
            break
        line = canonical[index]
        quantity, unit, food, notes = _vet_structure(
            line, (result or {}).get("ingredient") or {}, units
        )
        if notes:
            corrections.append({"line": line, "notes": notes})
        if unit or food:
            candidates[index] = _structured_item(base[index], line, quantity, unit, food)

    rejected: list[int] = []

    def apply(items: list[dict[str, Any]]) -> list[dict[str, Any]] | None:
        try:
            _request("PATCH", f"/api/recipes/{slug_or_id}", json={"recipeIngredient": items})
        except httpx.HTTPStatusError as exc:
            rejected.append(exc.response.status_code)
            return None
        return get_recipe(slug_or_id).get("recipeIngredient") or []

    stored = apply(candidates)
    if stored is None:
        # A rejected patch writes nothing, so the readable import is already intact.
        return {
            "parsed": False,
            "reason": f"Structured patch rejected by Mealie ({rejected[0]}); the recipe "
                      "was not modified",
            "canonical_lines": canonical,
        }

    # Mealie is the only authority on how a row renders, so compare what it stored.
    reverted: list[dict[str, Any]] = []
    final = list(candidates)
    for index, line in enumerate(canonical):
        if index >= len(stored):
            break
        shown = stored[index].get("display") or ""
        if not same_line(shown, line):
            final[index] = _unstructured_item(base[index], line)
            reverted.append({"line": line, "mealie_would_show": shown})

    if reverted:
        # The first patch succeeded, so these rows are stored reworded right now and
        # the revert is a repair, not a precaution — say so plainly if it fails.
        stored = apply(final)
        if stored is None:
            return {
                "parsed": False,
                "reason": f"Rows were reworded by Mealie and the repair patch failed "
                          f"({rejected[-1]}); these lines are still reworded in Mealie "
                          f"and need fixing by hand",
                "reworded_rows": reverted,
            }

    structured = [line for index, line in enumerate(canonical)
                  if final[index].get("unit") or final[index].get("food")]
    return {
        "parsed": True,
        "structured_rows": structured,
        "left_unstructured": reverted,
        "corrections": corrections,
        "lines_unchanged": all(same_line((stored[i].get("display") or ""), line)
                               for i, line in enumerate(canonical) if i < len(stored)),
        "recipe": get_recipe(slug_or_id),
    }


#: Hero image declared by a recipe page, in the order worth trusting.
_HERO_META = (
    re.compile(r'<meta[^>]+property=["\']og:image["\'][^>]+content=["\']([^"\']+)', re.I),
    re.compile(r'<meta[^>]+content=["\']([^"\']+)["\'][^>]+property=["\']og:image["\']', re.I),
    re.compile(r'<meta[^>]+name=["\']twitter:image["\'][^>]+content=["\']([^"\']+)', re.I),
)


def _hero_image_url(page_url: str) -> str | None:
    """Find the image a recipe page advertises as its own.

    Mealie's cover endpoint downloads the URL it is given and rejects anything
    that is not an image, so a page has to be reduced to a real image URL first.
    Fetched with a browser user-agent because recipe sites refuse default ones.
    """
    try:
        response = httpx.get(
            page_url,
            headers={"User-Agent": "Mozilla/5.0 (X11; Linux x86_64)"},
            timeout=30,
            follow_redirects=True,
        )
        response.raise_for_status()
    except httpx.HTTPError:
        return None
    for pattern in _HERO_META:
        match = pattern.search(response.text)
        if match:
            return html.unescape(match.group(1))
    return None


@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 from a URL or a local file.

    ``source_url`` may be a recipe page or a direct image URL. Mealie's endpoint
    only accepts the latter — handing it a page answers
    ``400 {"message": "Url is not an image"}`` — so a page is resolved to its
    ``og:image`` first and that is what gets sent.

    Local uploads must send the multipart ``extension`` field or the upload fails
    validation silently on this instance.
    """
    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:
        used_url = source_url
        method = "scrape"
        try:
            _request("POST", f"/api/recipes/{slug_or_id}/image", json={"url": used_url})
        except httpx.HTTPStatusError as exc:
            if exc.response.status_code != 400 or "not an image" not in exc.response.text:
                raise
            used_url = _hero_image_url(source_url)
            if not used_url:
                raise ValueError(
                    f"{source_url} is not an image and no og:image was found on the page; "
                    "pass a direct image URL or use image_path"
                ) from exc
            _request("POST", f"/api/recipes/{slug_or_id}/image", json={"url": used_url})
            method = "scrape (resolved og:image)"

    recipe = get_recipe(slug_or_id)
    result = {"method": method, "image": recipe.get("image"), "slug": recipe.get("slug")}
    if not image_path and used_url != source_url:
        result["resolved_image_url"] = used_url
    return result


# --- 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()


def _configure_read_only(
    server: MCPServer,
    write_tool_names: frozenset[str] = WRITE_TOOL_NAMES,
) -> None:
    """Remove mutating tools before exposing the server to a read-only client."""
    for name in sorted(write_tool_names):
        server.remove_tool(name)


def main_read_only() -> None:
    """Run the same server without import, patch, delete, parse, image, or HA writes."""
    _configure_read_only(mcp)
    mcp.run()