From e68b6a41fdaf7f33835e80632cb4522b4b7e0ee2 Mon Sep 17 00:00:00 2001 From: Oliver G Date: Sat, 21 Feb 2026 13:07:08 +0100 Subject: [PATCH] feat(wordpress): upload selected image and set featured_media on draft publish --- backend/app/wordpress.py | 87 +++++++++++++++++++++++++++++++++ backend/tests/test_wordpress.py | 63 ++++++++++++++++++++++++ 2 files changed, 150 insertions(+) create mode 100644 backend/tests/test_wordpress.py diff --git a/backend/app/wordpress.py b/backend/app/wordpress.py index adb4d9c..756a346 100644 --- a/backend/app/wordpress.py +++ b/backend/app/wordpress.py @@ -2,7 +2,10 @@ from __future__ import annotations import base64 import json +import mimetypes +from pathlib import Path from typing import Any +from urllib.parse import urlparse from urllib.request import Request, urlopen from .config import get_settings @@ -56,6 +59,77 @@ def _selected_image_url_from_meta(meta_json: str | None) -> str | None: return selected if isinstance(selected, str) and selected.strip() else None +def _download_image_bytes(url: str) -> tuple[bytes, str]: + req = Request( + url=url, + headers={ + "User-Agent": "rss-news-publisher/1.0", + "Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8", + }, + ) + with urlopen(req, timeout=20) as resp: + raw = resp.read() + content_type = resp.headers.get("Content-Type", "application/octet-stream") + if not content_type.lower().startswith("image/"): + raise RuntimeError(f"Ausgewählte Bild-URL liefert kein Bild ({content_type})") + return raw, content_type + + +def _guess_filename(image_url: str, content_type: str) -> str: + parsed = urlparse(image_url) + stem = Path(parsed.path).name or "article-image" + if "." not in stem: + ext = mimetypes.guess_extension(content_type.split(";")[0].strip()) or ".jpg" + stem = f"{stem}{ext}" + return stem + + +def _upload_featured_media( + *, + base_url: str, + auth_header: str, + image_url: str, + article_title: str, + source_url: str, +) -> int: + image_bytes, content_type = _download_image_bytes(image_url) + filename = _guess_filename(image_url, content_type) + + media_url = f"{base_url.rstrip('/')}/wp-json/wp/v2/media" + media_req = Request( + url=media_url, + data=image_bytes, + method="POST", + headers={ + "Authorization": auth_header, + "Content-Type": content_type, + "Content-Disposition": f'attachment; filename="{filename}"', + "Accept": "application/json", + "User-Agent": "rss-news-publisher/1.0", + }, + ) + with urlopen(media_req, timeout=30) as resp: + media_raw = resp.read().decode("utf-8", errors="replace") + media_payload = json.loads(media_raw) if media_raw else {} + media_id = int(media_payload.get("id", 0)) if isinstance(media_payload, dict) else 0 + if media_id <= 0: + raise RuntimeError(f"WordPress Media-Upload fehlgeschlagen: {media_payload}") + + # Optional metadata update for traceability. + _wp_request( + base_url=base_url, + auth_header=auth_header, + method="POST", + endpoint=f"media/{media_id}", + payload={ + "title": f"{article_title[:120]} - Bild", + "caption": f"Quelle: {source_url}", + "alt_text": article_title[:200], + }, + ) + return media_id + + def publish_article_draft(article: dict[str, Any]) -> tuple[int, str | None]: settings = get_settings() if not settings.wordpress_base_url or not settings.wordpress_username or not settings.wordpress_app_password: @@ -76,11 +150,24 @@ def publish_article_draft(article: dict[str, Any]) -> tuple[int, str | None]: footer += f"\n

Canonical: {canonical_url}

" content = f"{body}{footer}" + featured_media_id = None + selected_image_url = _selected_image_url_from_meta(article.get("meta_json")) + if selected_image_url: + featured_media_id = _upload_featured_media( + base_url=settings.wordpress_base_url, + auth_header=auth, + image_url=selected_image_url, + article_title=title, + source_url=source_url, + ) + payload = { "title": title, "content": content, "status": settings.wordpress_default_status, } + if featured_media_id: + payload["featured_media"] = featured_media_id wp_post_id = article.get("wp_post_id") if wp_post_id: diff --git a/backend/tests/test_wordpress.py b/backend/tests/test_wordpress.py new file mode 100644 index 0000000..f12c6e1 --- /dev/null +++ b/backend/tests/test_wordpress.py @@ -0,0 +1,63 @@ +import os +import unittest +from unittest.mock import patch + +from backend.app import config as config_module +from backend.app.wordpress import publish_article_draft + + +class TestWordpressPublish(unittest.TestCase): + def setUp(self) -> None: + os.environ["WORDPRESS_BASE_URL"] = "https://example.org" + os.environ["WORDPRESS_USERNAME"] = "wp-user" + os.environ["WORDPRESS_APP_PASSWORD"] = "wp-pass" + config_module.get_settings.cache_clear() + + def tearDown(self) -> None: + for key in ("WORDPRESS_BASE_URL", "WORDPRESS_USERNAME", "WORDPRESS_APP_PASSWORD"): + os.environ.pop(key, None) + config_module.get_settings.cache_clear() + + @patch("backend.app.wordpress._upload_featured_media") + @patch("backend.app.wordpress._wp_request") + def test_publish_sets_featured_media_when_selected_image_exists(self, mock_wp_request, mock_upload_media) -> None: + mock_upload_media.return_value = 456 + mock_wp_request.return_value = {"id": 321, "link": "https://example.org/?p=321"} + + article = { + "title": "Testartikel", + "content_raw": "Inhalt", + "source_url": "https://example.com/source", + "canonical_url": "https://example.com/source", + "meta_json": '{"image_review":{"selected_url":"https://example.com/image.jpg"}}', + } + post_id, post_url = publish_article_draft(article) + + self.assertEqual(post_id, 321) + self.assertIn("?p=321", post_url or "") + self.assertTrue(mock_upload_media.called) + payload = mock_wp_request.call_args.kwargs["payload"] + self.assertEqual(payload.get("featured_media"), 456) + + @patch("backend.app.wordpress._upload_featured_media") + @patch("backend.app.wordpress._wp_request") + def test_publish_without_selected_image_has_no_featured_media(self, mock_wp_request, mock_upload_media) -> None: + mock_wp_request.return_value = {"id": 654, "link": "https://example.org/?p=654"} + + article = { + "title": "Testartikel", + "content_raw": "Inhalt", + "source_url": "https://example.com/source", + "canonical_url": "https://example.com/source", + "meta_json": "{}", + } + post_id, _ = publish_article_draft(article) + + self.assertEqual(post_id, 654) + self.assertFalse(mock_upload_media.called) + payload = mock_wp_request.call_args.kwargs["payload"] + self.assertNotIn("featured_media", payload) + + +if __name__ == "__main__": + unittest.main()