feat(wordpress): only assign tags that are established

_resolve_wp_tag_ids created a WordPress tag for every keyword the rewriter
invented, up to 12 per post. That is where the 3.055 tags for 955 posts
came from - 1.683 of them used exactly once, 473 attached to no post at
all. The categories are stable now, but the tags would simply grow back.

Three changes, all on the write path:

A proposed tag has to appear for wordpress_new_tag_min_proposals (3)
different articles before it is created. Proposals are counted in the new
tag_proposals table, keyed on (name, article), so re-publishing an article
does not inflate its own count. Tags that already exist in WordPress are
assigned as before - the gate only guards creation.

Only the first wordpress_max_tags_per_post (5) tags reach WordPress. The
full list still feeds the category rules, which were validated against it.

The lookup fallback of reusing the first search hit is gone. It filed
"Camping" under the unrelated existing tag "Campingplatz" whenever the
exact tag was missing, which quietly produced wrong tags rather than none.

If the proposal bookkeeping fails, nothing is creatable that run: existing
tags still get assigned and the taxonomy stays put, rather than falling
back to creating everything.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Oliver 2026-07-31 08:21:53 +02:00
parent f0d10112ad
commit 7163b1bd7e
No known key found for this signature in database
5 changed files with 266 additions and 15 deletions

View file

@ -1,9 +1,12 @@
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from backend.app import config as config_module
from backend.app import wordpress as wordpress_module
from backend.app.db import init_db
from backend.app.wordpress import publish_article_draft
@ -12,15 +15,28 @@ class TestWordpressPublish(unittest.TestCase):
os.environ["WORDPRESS_BASE_URL"] = "https://example.org"
os.environ["WORDPRESS_USERNAME"] = "wp-user"
os.environ["WORDPRESS_APP_PASSWORD"] = "wp-pass"
# Publishing records tag proposals, so it needs a database of its own -
# otherwise the tests would write into the live one.
self.tmp_dir = tempfile.TemporaryDirectory()
os.environ["APP_DB_PATH"] = str(Path(self.tmp_dir.name) / "test.db")
config_module.get_settings.cache_clear()
init_db()
# The category lookup is cached for the process lifetime; without this
# one test would resolve a slug that the next one expects to be missing.
wordpress_module._category_id_cache.clear()
def tearDown(self) -> None:
for key in ("WORDPRESS_BASE_URL", "WORDPRESS_USERNAME", "WORDPRESS_APP_PASSWORD"):
for key in (
"WORDPRESS_BASE_URL",
"WORDPRESS_USERNAME",
"WORDPRESS_APP_PASSWORD",
"APP_DB_PATH",
"WORDPRESS_MAX_TAGS_PER_POST",
"WORDPRESS_NEW_TAG_MIN_PROPOSALS",
):
os.environ.pop(key, None)
config_module.get_settings.cache_clear()
self.tmp_dir.cleanup()
@patch("backend.app.wordpress._upload_featured_media")
@patch("backend.app.wordpress._wp_request")
@ -87,7 +103,7 @@ class TestWordpressPublish(unittest.TestCase):
@patch("backend.app.wordpress._upload_featured_media")
@patch("backend.app.wordpress._wp_request")
def test_publish_resolves_and_sets_tags(self, mock_wp_request, mock_upload_media) -> None:
def test_publish_resolves_existing_tags_and_skips_unknown_ones(self, mock_wp_request, mock_upload_media) -> None:
def _fake_wp_request(**kwargs):
endpoint = kwargs.get("endpoint", "")
method = kwargs.get("method", "")
@ -95,17 +111,13 @@ class TestWordpressPublish(unittest.TestCase):
if "Rheingas" in endpoint:
return [{"id": 11, "name": "Rheingas"}]
return []
if method == "POST" and endpoint == "tags":
name = (kwargs.get("payload") or {}).get("name")
if name == "Gasflasche":
return {"id": 12, "name": "Gasflasche"}
return {"id": 13, "name": str(name)}
if method == "POST" and endpoint == "posts":
return {"id": 900, "link": "https://example.org/?p=900"}
return {}
mock_wp_request.side_effect = _fake_wp_request
article = {
"id": 1,
"title": "Tag Test",
"content_raw": "Inhalt",
"source_url": "https://example.com/source",
@ -117,7 +129,138 @@ class TestWordpressPublish(unittest.TestCase):
post_calls = [call for call in mock_wp_request.call_args_list if call.kwargs.get("endpoint") == "posts"]
self.assertEqual(len(post_calls), 1)
payload = post_calls[0].kwargs.get("payload", {})
self.assertEqual(payload.get("tags"), [11, 12])
# "Gasflasche" was proposed for the first time and is not created yet.
self.assertEqual(payload.get("tags"), [11])
self.assertFalse(
any(c.kwargs.get("method") == "POST" and c.kwargs.get("endpoint") == "tags"
for c in mock_wp_request.call_args_list)
)
@patch("backend.app.wordpress._upload_featured_media")
@patch("backend.app.wordpress._wp_request")
def test_new_tag_is_created_once_enough_articles_proposed_it(self, mock_wp_request, mock_upload_media) -> None:
created_names: list[str] = []
def _fake_wp_request(**kwargs):
endpoint = kwargs.get("endpoint", "")
method = kwargs.get("method", "")
if method == "GET" and endpoint.startswith("tags?search="):
return []
if method == "POST" and endpoint == "tags":
created_names.append(str((kwargs.get("payload") or {}).get("name")))
return {"id": 55, "name": "Wintercamping"}
if method == "POST" and endpoint == "posts":
return {"id": 910, "link": "https://example.org/?p=910"}
return {}
mock_wp_request.side_effect = _fake_wp_request
def _article(article_id: int) -> dict:
return {
"id": article_id,
"title": f"Artikel {article_id}",
"content_raw": "Inhalt",
"source_url": f"https://example.com/source/{article_id}",
"canonical_url": f"https://example.com/source/{article_id}",
"meta_json": '{"generated_tags":["Wintercamping"]}',
}
for article_id in (1, 2):
publish_article_draft(_article(article_id))
self.assertEqual(created_names, [])
publish_article_draft(_article(3))
self.assertEqual(created_names, ["Wintercamping"])
payload = [c for c in mock_wp_request.call_args_list if c.kwargs.get("endpoint") == "posts"][-1].kwargs["payload"]
self.assertEqual(payload.get("tags"), [55])
@patch("backend.app.wordpress._upload_featured_media")
@patch("backend.app.wordpress._wp_request")
def test_republishing_the_same_article_does_not_count_twice(self, mock_wp_request, mock_upload_media) -> None:
def _fake_wp_request(**kwargs):
endpoint = kwargs.get("endpoint", "")
method = kwargs.get("method", "")
if method == "GET" and endpoint.startswith("tags?search="):
return []
if method == "POST" and endpoint == "tags":
return {"id": 66, "name": "Solaranlage"}
if method == "POST":
return {"id": 920, "link": "https://example.org/?p=920"}
return {}
mock_wp_request.side_effect = _fake_wp_request
article = {
"id": 7,
"title": "Immer wieder derselbe Artikel",
"content_raw": "Inhalt",
"source_url": "https://example.com/source",
"canonical_url": "https://example.com/source",
"meta_json": '{"generated_tags":["Solaranlage"]}',
}
for _ in range(4):
publish_article_draft(article)
self.assertFalse(
any(c.kwargs.get("method") == "POST" and c.kwargs.get("endpoint") == "tags"
for c in mock_wp_request.call_args_list)
)
@patch("backend.app.wordpress._upload_featured_media")
@patch("backend.app.wordpress._wp_request")
def test_publish_caps_the_number_of_tags_per_post(self, mock_wp_request, mock_upload_media) -> None:
looked_up: list[str] = []
def _fake_wp_request(**kwargs):
endpoint = kwargs.get("endpoint", "")
method = kwargs.get("method", "")
if method == "GET" and endpoint.startswith("tags?search="):
looked_up.append(endpoint)
return []
if method == "POST" and endpoint == "posts":
return {"id": 930, "link": "https://example.org/?p=930"}
return {}
mock_wp_request.side_effect = _fake_wp_request
names = [f"Tag{i}" for i in range(1, 13)]
article = {
"id": 42,
"title": "Viele Schlagwörter",
"content_raw": "Inhalt",
"source_url": "https://example.com/source",
"canonical_url": "https://example.com/source",
"meta_json": '{"generated_tags":' + str(names).replace("'", '"') + "}",
}
publish_article_draft(article)
self.assertEqual(len(looked_up), 5)
@patch("backend.app.wordpress._upload_featured_media")
@patch("backend.app.wordpress._wp_request")
def test_tag_lookup_ignores_near_misses(self, mock_wp_request, mock_upload_media) -> None:
"""A search hit that is not the exact name must not be reused.
Plain substring reuse filed "Camping" under the unrelated existing tag
"Campingplatz" whenever the exact tag was missing.
"""
def _fake_wp_request(**kwargs):
endpoint = kwargs.get("endpoint", "")
method = kwargs.get("method", "")
if method == "GET" and endpoint.startswith("tags?search="):
return [{"id": 99, "name": "Campingplatz"}]
if method == "POST" and endpoint == "posts":
return {"id": 940, "link": "https://example.org/?p=940"}
return {}
mock_wp_request.side_effect = _fake_wp_request
article = {
"id": 5,
"title": "Naher Treffer",
"content_raw": "Inhalt",
"source_url": "https://example.com/source",
"canonical_url": "https://example.com/source",
"meta_json": '{"generated_tags":["Camping"]}',
}
publish_article_draft(article)
payload = [c for c in mock_wp_request.call_args_list if c.kwargs.get("endpoint") == "posts"][0].kwargs["payload"]
self.assertNotIn("tags", payload)
@patch("backend.app.wordpress._upload_featured_media")
@patch("backend.app.wordpress._wp_request")