Resolve recipe pages to their hero image for cover scraping
Mealie's POST /api/recipes/{slug}/image downloads the URL it is given and
rejects anything that is not an image, so handing it a recipe page answered
400 {"message": "Url is not an image"}. This was read as a source-specific
failure on ica.se, but that page's own og:image uploads fine (verified live,
200 + 74608 bytes of image/webp), and httpbin's JPEG works too — the endpoint
simply never accepted pages.
Callers naturally pass the recipe page they just imported from, so resolve a
page to its advertised og:image and retry rather than making every caller know
the distinction. A page with no discoverable image now raises instead of
leaving the recipe silently coverless.
Co-authored-by: fredamn76 <fredrik.fallman@gmail.com>
Signed-off-by: fredamn76 <fredrik.fallman@gmail.com>
This commit is contained in:
+58
-5
@@ -1,6 +1,7 @@
|
||||
"""Small, confirmation-friendly MCP facade over the Mealie API."""
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
@@ -546,17 +547,53 @@ def parse_ingredients(slug_or_id: str) -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
#: 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 by scraping a URL or uploading a local file.
|
||||
"""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. When a scrape does not produce a visible
|
||||
cover (observed on Allrecipes), download the hero image and upload it instead.
|
||||
validation silently on this instance.
|
||||
"""
|
||||
if not source_url and not image_path:
|
||||
raise ValueError("Provide either source_url or image_path")
|
||||
@@ -575,11 +612,27 @@ def set_cover_image(
|
||||
_raise_for_status(response)
|
||||
method = f"upload ({extension})"
|
||||
else:
|
||||
_request("POST", f"/api/recipes/{slug_or_id}/image", json={"url": source_url})
|
||||
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)
|
||||
return {"method": method, "image": recipe.get("image"), "slug": recipe.get("slug")}
|
||||
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 -----------------------------------------------------------------
|
||||
|
||||
@@ -259,3 +259,92 @@ class TestCoverImage:
|
||||
def test_requires_a_source(self):
|
||||
with pytest.raises(ValueError):
|
||||
server.set_cover_image("x")
|
||||
|
||||
def test_a_recipe_page_is_resolved_to_its_hero_image(self, mock_mealie, monkeypatch):
|
||||
# Verified live: Mealie answers 400 "Url is not an image" for a recipe page,
|
||||
# and 200 for that same page's og:image.
|
||||
page = "https://www.ica.se/recept/kalsoppa-722069/"
|
||||
hero = "https://assets.icanet.se/imagevaultfiles/kalsoppa.jpg"
|
||||
posted: list[str] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
import json as _json
|
||||
|
||||
if request.method == "POST" and request.url.path.endswith("/image"):
|
||||
url = _json.loads(request.content)["url"]
|
||||
posted.append(url)
|
||||
if url == page:
|
||||
return httpx.Response(
|
||||
400, json={"detail": {"message": "Url is not an image", "error": True}}
|
||||
)
|
||||
return httpx.Response(200, json=None)
|
||||
return httpx.Response(200, json={"slug": "soppa", "image": "abcd"})
|
||||
|
||||
mock_mealie(handler)
|
||||
monkeypatch.setattr(
|
||||
server, "_hero_image_url", lambda url: hero if url == page else None
|
||||
)
|
||||
|
||||
result = server.set_cover_image("soppa", source_url=page)
|
||||
|
||||
# The page is tried first, then retried with the image it advertises.
|
||||
assert posted == [page, hero]
|
||||
assert result["method"] == "scrape (resolved og:image)"
|
||||
assert result["resolved_image_url"] == hero
|
||||
|
||||
def test_a_direct_image_url_is_sent_unchanged(self, mock_mealie, monkeypatch):
|
||||
recorded = mock_mealie(lambda r: httpx.Response(200, json={"slug": "x", "image": "img"}))
|
||||
monkeypatch.setattr(server, "_hero_image_url", lambda url: pytest.fail("not needed"))
|
||||
|
||||
result = server.set_cover_image("x", source_url="https://example.com/hero.jpg")
|
||||
|
||||
assert result["method"] == "scrape"
|
||||
assert "resolved_image_url" not in result
|
||||
assert any(r.method == "POST" for r in recorded)
|
||||
|
||||
def test_a_page_with_no_hero_image_fails_loudly(self, mock_mealie, monkeypatch):
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.method == "POST" and request.url.path.endswith("/image"):
|
||||
return httpx.Response(400, json={"detail": {"message": "Url is not an image"}})
|
||||
return httpx.Response(200, json={"slug": "x"})
|
||||
|
||||
mock_mealie(handler)
|
||||
monkeypatch.setattr(server, "_hero_image_url", lambda url: None)
|
||||
|
||||
with pytest.raises(ValueError, match="no og:image"):
|
||||
server.set_cover_image("x", source_url="https://example.com/page")
|
||||
|
||||
|
||||
class TestHeroImageUrl:
|
||||
def test_reads_og_image_in_either_attribute_order(self, monkeypatch):
|
||||
pages = {
|
||||
"a": '<meta property="og:image" content="https://x.test/a.jpg">',
|
||||
"b": '<meta content="https://x.test/b.jpg" property="og:image">',
|
||||
"c": '<meta name="twitter:image" content="https://x.test/c.jpg">',
|
||||
"d": "<html><body>no image here</body></html>",
|
||||
}
|
||||
|
||||
def fake_get(url, **kwargs):
|
||||
return httpx.Response(200, text=pages[url], request=httpx.Request("GET", "http://t"))
|
||||
|
||||
monkeypatch.setattr(server.httpx, "get", fake_get)
|
||||
assert server._hero_image_url("a") == "https://x.test/a.jpg"
|
||||
assert server._hero_image_url("b") == "https://x.test/b.jpg"
|
||||
assert server._hero_image_url("c") == "https://x.test/c.jpg"
|
||||
assert server._hero_image_url("d") is None
|
||||
|
||||
def test_unescapes_entities_in_the_url(self, monkeypatch):
|
||||
# Query strings in og:image arrive HTML-escaped and must be decoded.
|
||||
page = '<meta property="og:image" content="https://x.test/a.jpg?w=1&h=2">'
|
||||
monkeypatch.setattr(
|
||||
server.httpx, "get",
|
||||
lambda url, **kw: httpx.Response(200, text=page, request=httpx.Request("GET", "http://t")),
|
||||
)
|
||||
assert server._hero_image_url("p") == "https://x.test/a.jpg?w=1&h=2"
|
||||
|
||||
def test_an_unreachable_page_is_not_an_error(self, monkeypatch):
|
||||
def boom(url, **kwargs):
|
||||
raise httpx.ConnectError("no route")
|
||||
|
||||
monkeypatch.setattr(server.httpx, "get", boom)
|
||||
assert server._hero_image_url("https://example.com/page") is None
|
||||
|
||||
Reference in New Issue
Block a user