From 077c5db01bf09ac263dc20aa42f23589dc4592f9 Mon Sep 17 00:00:00 2001 From: Jens <1742418+1cu@users.noreply.github.com> Date: Sat, 4 Jul 2026 19:47:13 +0200 Subject: [PATCH] fix: prevent ambiguous title deletes (#1202) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## ℹ️ Description - Link to the related issue(s): Issue # - Prevent title-based deletion from deleting multiple same-title remote ads. - Make explicit ID deletion target only the configured ID, while ID-less title matching now fails closed on ambiguous matches. ## 📋 Changes Summary - Split delete selection so configured IDs are exact-ID only and title matching is only used for ID-less ads. - Require strict published-ad fetching when ID-less title matching may run, so ambiguity detection uses a complete list. - Skip ambiguous title matches before browser delete requests, ID mutation, or after-delete cleanup. - Added German translation for the new skip message. - Updated config/docs/generated schema wording for `delete_old_ads_by_title`. - Added unit coverage for ambiguous title fail-closed behavior, exact-ID behavior, strict fetch selection, and exact delete URLs. ### ⚙️ Type of Change Select the type(s) of change(s) included in this pull request: - [x] 🐞 Bug fix (non-breaking change which fixes an issue) - [ ] ✨ New feature (adds new functionality without breaking existing usage) - [ ] 💥 Breaking change (changes that might break existing user setups, scripts, or configurations) ## ✅ Checklist Before requesting a review, confirm the following: - [x] I have reviewed my changes to ensure they meet the project's standards. - [x] I have tested my changes and ensured that all tests pass (`pdm run test`). - [x] I have formatted the code (`pdm run format`). - [x] I have verified that linting passes (`pdm run lint`). - [x] I have updated documentation where necessary. By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice. Validation: - `pdm run generate-schemas` - `pdm run generate-config` - `pdm run pytest tests/unit/test_delete_flow.py` - `pdm run format` - `pdm run lint` - `pdm run test` ## Summary by CodeRabbit * **Bug Fixes** * Improved old-ad deletion behavior for items without an ID, including title-based matching only when needed. * Made title-based deletion fail closed: ambiguous title matches are now skipped. * Ensured exact-ID deletions target only the specified ID and don’t expand to same-title ads. * Skips title-based matching for a batch when published-ad retrieval can’t be completed, while continuing exact-ID deletions. * **Documentation** * Updated configuration documentation for title-based old-ad deletion to reflect the new matching and skip behavior. --- docs/CONFIGURATION.md | 2 +- docs/config.default.yaml | 2 +- schemas/config.schema.json | 2 +- src/kleinanzeigen_bot/delete_flow.py | 39 ++++- src/kleinanzeigen_bot/model/config_model.py | 5 +- .../resources/translations.de.yaml | 2 + tests/unit/test_delete_flow.py | 154 +++++++++++++++--- 7 files changed, 174 insertions(+), 32 deletions(-) diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index eb317a4..aee3059 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -272,7 +272,7 @@ Publishing configuration. ```yaml publishing: delete_old_ads: "AFTER_PUBLISH" # one of: AFTER_PUBLISH, BEFORE_PUBLISH, NEVER - delete_old_ads_by_title: true # only works if delete_old_ads is set to BEFORE_PUBLISH + delete_old_ads_by_title: true # match by title before publish or for ID-less deletes; ambiguous matches are skipped ``` ### captcha diff --git a/docs/config.default.yaml b/docs/config.default.yaml index 6df683f..4ffcb57 100644 --- a/docs/config.default.yaml +++ b/docs/config.default.yaml @@ -171,7 +171,7 @@ publishing: # • NEVER delete_old_ads: AFTER_PUBLISH - # match old ads by title when deleting (only works with BEFORE_PUBLISH) + # match ads by title when deleting old ads before publish or deleting ID-less ads; ambiguous title matches are skipped delete_old_ads_by_title: true # local file and folder rename behavior after a successful publish changes the ad ID. When TEMPLATE_MATCH is enabled, the download.folder_name_template and download.ad_file_name_template are used to determine which paths qualify for renaming — only paths whose names match the template structure are updated. diff --git a/schemas/config.schema.json b/schemas/config.schema.json index b2edf9a..523ea33 100644 --- a/schemas/config.schema.json +++ b/schemas/config.schema.json @@ -707,7 +707,7 @@ }, "delete_old_ads_by_title": { "default": true, - "description": "match old ads by title when deleting (only works with BEFORE_PUBLISH)", + "description": "match ads by title when deleting old ads before publish or deleting ID-less ads; ambiguous title matches are skipped", "title": "Delete Old Ads By Title", "type": "boolean" }, diff --git a/src/kleinanzeigen_bot/delete_flow.py b/src/kleinanzeigen_bot/delete_flow.py index 5c62dc7..00f3d1f 100644 --- a/src/kleinanzeigen_bot/delete_flow.py +++ b/src/kleinanzeigen_bot/delete_flow.py @@ -44,13 +44,29 @@ async def delete_ads( count = 0 deleted_count = 0 - published_ads_list = await published_ads.fetch_published_ads(web, root_url) + needs_title_matching = delete_old_ads_by_title and any(ad_cfg.id is None for _, ad_cfg, _ in ad_cfgs) + title_matching_fetch_error:published_ads.PublishedAdsFetchIncompleteError | None = None + if needs_title_matching: + try: + published_ads_list = await published_ads.fetch_published_ads(web, root_url, strict = True) + except published_ads.PublishedAdsFetchIncompleteError as ex: + published_ads_list = [] + title_matching_fetch_error = ex + else: + published_ads_list = [] for ad_file, ad_cfg, ad_cfg_orig in ad_cfgs: count += 1 LOG.info("Processing %s/%s: '%s' from [%s]...", count, len(ad_cfgs), ad_cfg.title, ad_file) - result = await delete_ad(web, root_url, ad_cfg, published_ads_list, delete_old_ads_by_title = delete_old_ads_by_title) + if ad_cfg.id is None and delete_old_ads_by_title and title_matching_fetch_error is not None: + LOG.error( + " -> SKIPPED: title-based deletion requires a complete published ads list: %s", + title_matching_fetch_error, + ) + result = DeleteResult(deleted = False, attempted = False) + else: + result = await delete_ad(web, root_url, ad_cfg, published_ads_list, delete_old_ads_by_title = delete_old_ads_by_title) if result.deleted: deleted_count += 1 @@ -85,10 +101,13 @@ async def delete_ad( """ LOG.info("Deleting ad '%s' if already present...", ad_cfg.title) - # Phase A: Build set of IDs to delete, unified across both modes + # Phase A: Build set of IDs to delete. Explicit IDs are exact-ID only; + # title matching is only used for ID-less ads and must fail closed when ambiguous. ids_to_delete:set[int] = set() - if delete_old_ads_by_title: + if ad_cfg.id is not None: + ids_to_delete.add(ad_cfg.id) + elif delete_old_ads_by_title: for published_ad in published_ads_list: raw_id = published_ad.get("id") if raw_id is None: @@ -100,11 +119,17 @@ async def delete_ad( LOG.debug("Skipping published ad with invalid id: %r", raw_id) continue published_ad_title = published_ad.get("title", "") - if ad_cfg.id == published_ad_id or ad_cfg.title == published_ad_title: + if ad_cfg.title == published_ad_title: LOG.debug(" -> matched ad %s '%s' for deletion", published_ad_id, published_ad_title) ids_to_delete.add(published_ad_id) - elif ad_cfg.id is not None: - ids_to_delete.add(ad_cfg.id) + + if len(ids_to_delete) > 1: + LOG.error( + " -> SKIPPED: title '%s' matched multiple published ads (%s); delete by ID instead", + ad_cfg.title, + ", ".join(str(ad_id) for ad_id in sorted(ids_to_delete)), + ) + return DeleteResult(deleted = False, attempted = False) # Early return if nothing to delete — skip page open, CSRF fetch, and sleep if not ids_to_delete: diff --git a/src/kleinanzeigen_bot/model/config_model.py b/src/kleinanzeigen_bot/model/config_model.py index 9f5d333..30516c0 100644 --- a/src/kleinanzeigen_bot/model/config_model.py +++ b/src/kleinanzeigen_bot/model/config_model.py @@ -266,7 +266,10 @@ class PublishingConfig(ContextualModel): delete_old_ads:Literal["BEFORE_PUBLISH", "AFTER_PUBLISH", "NEVER"] | None = Field( default = "AFTER_PUBLISH", description = "when to delete old versions of republished ads", examples = ["BEFORE_PUBLISH", "AFTER_PUBLISH", "NEVER"] ) - delete_old_ads_by_title:bool = Field(default = True, description = "match old ads by title when deleting (only works with BEFORE_PUBLISH)") + delete_old_ads_by_title:bool = Field( + default = True, + description = "match ads by title when deleting old ads before publish or deleting ID-less ads; ambiguous title matches are skipped", + ) local_path_renaming:LocalPathRenamingConfig = Field( default_factory = LocalPathRenamingConfig, description = ( diff --git a/src/kleinanzeigen_bot/resources/translations.de.yaml b/src/kleinanzeigen_bot/resources/translations.de.yaml index 50925e2..b67ac87 100644 --- a/src/kleinanzeigen_bot/resources/translations.de.yaml +++ b/src/kleinanzeigen_bot/resources/translations.de.yaml @@ -378,11 +378,13 @@ kleinanzeigen_bot/delete_flow.py: "Processing %s/%s: '%s' from [%s]...": "Verarbeite %s/%s: '%s' von [%s]..." "DONE: Deleted %s of %s": "FERTIG: %s von %s gelöscht" "ad": "Anzeige" + " -> SKIPPED: title-based deletion requires a complete published ads list: %s": " -> ÜBERSPRUNGEN: titelbasierte Löschung erfordert eine vollständige Liste veröffentlichter Anzeigen: %s" delete_ad: "Deleting ad '%s' if already present...": "Lösche Anzeige '%s', falls bereits vorhanden..." "Expected CSRF Token not found in HTML content!": "Erwartetes CSRF-Token wurde im HTML-Inhalt nicht gefunden!" " -> SKIPPED: no published ad matched '%s' for deletion": " -> ÜBERSPRUNGEN: Keine veröffentlichte Anzeige '%s' zum Löschen gefunden" + " -> SKIPPED: title '%s' matched multiple published ads (%s); delete by ID instead": " -> ÜBERSPRUNGEN: Titel '%s' passt zu mehreren veröffentlichten Anzeigen (%s); stattdessen per ID löschen" " -> SUCCESS: deleted ad '%s' (ID: %s)": " -> ERFOLG: Anzeige '%s' (ID: %s) gelöscht" " -> ad %s not found (status %s), may have been removed already": " -> Anzeige %s nicht gefunden (Status %s), möglicherweise bereits entfernt" diff --git a/tests/unit/test_delete_flow.py b/tests/unit/test_delete_flow.py index 4df0b50..5e34e58 100644 --- a/tests/unit/test_delete_flow.py +++ b/tests/unit/test_delete_flow.py @@ -14,6 +14,7 @@ from kleinanzeigen_bot import delete_flow from kleinanzeigen_bot.app import KleinanzeigenBot from kleinanzeigen_bot.delete_flow import DeleteResult from kleinanzeigen_bot.model.ad_model import Ad +from kleinanzeigen_bot.published_ads import PublishedAdsFetchIncompleteError def remove_fields(config:dict[str, Any], *fields:str) -> dict[str, Any]: @@ -103,7 +104,7 @@ class TestKleinanzeigenBotAdDeletion: minimal_ad_config:dict[str, Any], ) -> None: """When no published ads match, should return False without opening any pages.""" - ad_cfg = Ad.model_validate(minimal_ad_config | {"title": "No Match Title", "id": 99999}) + ad_cfg = Ad.model_validate(minimal_ad_config | {"title": "No Match Title", "id": None}) published_ads = [{"title": "Different Title", "id": 12345}] with ( @@ -121,7 +122,7 @@ class TestKleinanzeigenBotAdDeletion: assert isinstance(result, DeleteResult) assert result == DeleteResult(deleted = False, attempted = False) - assert ad_cfg.id == 99999 # Preserved — no deletion attempted + assert ad_cfg.id is None # Preserved — no deletion attempted mock_web_open.assert_not_called() mock_web_find.assert_not_called() mock_web_sleep.assert_not_called() @@ -208,8 +209,10 @@ class TestKleinanzeigenBotAdDeletion: assert "ids=0" in mock_request.call_args[1]["url"] @pytest.mark.asyncio - async def test_delete_ad_multiple_title_matches(self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any]) -> None: - """When multiple published ads match by title with mixed 200/404, should return True.""" + async def test_delete_ad_multiple_title_matches_fails_closed( + self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any], + ) -> None: + """When multiple published ads match by title, should skip deletion as ambiguous.""" ad_cfg = Ad.model_validate(minimal_ad_config | {"title": "Test Title", "id": None}) published_ads = [ {"title": "Test Title", "id": 100}, @@ -217,19 +220,12 @@ class TestKleinanzeigenBotAdDeletion: {"title": "Test Title", "id": 300}, ] - response_sequence = [ - {"statusCode": 200, "statusMessage": "OK", "content": "{}"}, - {"statusCode": 404, "statusMessage": "Not Found", "content": "{}"}, - {"statusCode": 200, "statusMessage": "OK", "content": "{}"}, - ] - with ( - patch.object(test_bot, "web_open", new_callable = AsyncMock), - patch.object(test_bot, "web_find", new_callable = AsyncMock) as mock_find, - patch.object(test_bot, "web_sleep", new_callable = AsyncMock), - patch.object(test_bot, "web_request", new_callable = AsyncMock, side_effect = response_sequence) as mock_request, + patch.object(test_bot, "web_open", new_callable = AsyncMock) as mock_web_open, + patch.object(test_bot, "web_find", new_callable = AsyncMock) as mock_web_find, + patch.object(test_bot, "web_sleep", new_callable = AsyncMock) as mock_web_sleep, + patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, ): - mock_find.return_value.attrs = {"content": "some-token"} result = await delete_flow.delete_ad( web = test_bot, root_url = test_bot.root_url, ad_cfg = ad_cfg, @@ -238,15 +234,24 @@ class TestKleinanzeigenBotAdDeletion: ) assert isinstance(result, DeleteResult) - assert result == DeleteResult(deleted = True, attempted = True) + assert result == DeleteResult(deleted = False, attempted = False) assert ad_cfg.id is None - assert mock_request.call_count == 3 + mock_web_open.assert_not_called() + mock_web_find.assert_not_called() + mock_web_sleep.assert_not_called() + mock_request.assert_not_called() @pytest.mark.asyncio - async def test_delete_ad_title_and_id_match_deduplicated(self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any]) -> None: - """When the same ad matches by both title and ID, it should be deleted only once.""" + async def test_delete_ad_with_id_does_not_expand_to_same_title_matches( + self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any], + ) -> None: + """When an ID is present, title matching must not expand deletion to other ads.""" ad_cfg = Ad.model_validate(minimal_ad_config | {"title": "Test Title", "id": 100}) - published_ads = [{"title": "Test Title", "id": 100}] + published_ads = [ + {"title": "Test Title", "id": 100}, + {"title": "Test Title", "id": 200}, + {"title": "Test Title", "id": 300}, + ] ok_response = {"statusCode": 200, "statusMessage": "OK", "content": "{}"} with ( @@ -266,7 +271,37 @@ class TestKleinanzeigenBotAdDeletion: assert isinstance(result, DeleteResult) assert result == DeleteResult(deleted = True, attempted = True) assert ad_cfg.id is None - mock_request.assert_called_once() # Deduplicated — only one request + mock_request.assert_called_once() + assert mock_request.call_args.kwargs["url"] == f"{test_bot.root_url}/m-anzeigen-loeschen.json?ids=100" + + @pytest.mark.asyncio + async def test_delete_ad_with_id_ignores_published_ads_list_for_exact_id_delete( + self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any], + ) -> None: + """When an ID is present, deletion should target that exact ID even without a published-list match.""" + ad_cfg = Ad.model_validate(minimal_ad_config | {"title": "Test Title", "id": 100}) + published_ads = [{"title": "Other Title", "id": 200}] + ok_response = {"statusCode": 200, "statusMessage": "OK", "content": "{}"} + + with ( + patch.object(test_bot, "web_open", new_callable = AsyncMock), + patch.object(test_bot, "web_find", new_callable = AsyncMock) as mock_find, + patch.object(test_bot, "web_sleep", new_callable = AsyncMock), + patch.object(test_bot, "web_request", new_callable = AsyncMock, return_value = ok_response) as mock_request, + ): + mock_find.return_value.attrs = {"content": "some-token"} + result = await delete_flow.delete_ad( + web = test_bot, root_url = test_bot.root_url, + ad_cfg = ad_cfg, + published_ads_list = published_ads, + delete_old_ads_by_title = True, + ) + + assert isinstance(result, DeleteResult) + assert result == DeleteResult(deleted = True, attempted = True) + assert ad_cfg.id is None + mock_request.assert_called_once() + assert mock_request.call_args.kwargs["url"] == f"{test_bot.root_url}/m-anzeigen-loeschen.json?ids=100" @pytest.mark.asyncio async def test_delete_ad_exception_preserves_id(self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any]) -> None: @@ -395,6 +430,83 @@ class TestDeleteAdsAfterDeletePolicy: # delete_ad called twice (once per ad) assert mock_delete_ad.call_count == 2 + @pytest.mark.asyncio + async def test_delete_ads_fetches_published_ads_strictly_for_id_less_title_matching( + self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any], tmp_path:Path, + ) -> None: + """Title matching needs a complete published-ad list to detect ambiguous matches safely.""" + test_bot.config.deleting.after_delete = "NONE" + ad_file, ad_cfg, ad_cfg_orig = self._make_ad(minimal_ad_config, tmp_path) + ad_cfg.id = None + ad_cfg_orig.pop("id", None) + + with ( + patch("kleinanzeigen_bot.published_ads.fetch_published_ads", new_callable = AsyncMock, return_value = []) as mock_fetch, + patch("kleinanzeigen_bot.delete_flow.delete_ad", new_callable = AsyncMock, return_value = DeleteResult(deleted = False, attempted = False)), + patch.object(test_bot, "web_sleep", new_callable = AsyncMock), + ): + await delete_flow.delete_ads( + web = test_bot, root_url = test_bot.root_url, + after_delete = test_bot.config.deleting.after_delete, + delete_old_ads_by_title = True, + ad_cfgs = [(ad_file, ad_cfg, ad_cfg_orig)], + ) + + mock_fetch.assert_awaited_once_with(test_bot, test_bot.root_url, strict = True) + + @pytest.mark.asyncio + async def test_delete_ads_skips_published_ads_fetch_for_id_only_deletes( + self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any], tmp_path:Path, + ) -> None: + """ID-only deletes should not depend on fetching unrelated published-ad pages.""" + test_bot.config.deleting.after_delete = "NONE" + ad_file, ad_cfg, ad_cfg_orig = self._make_ad(minimal_ad_config, tmp_path) + + with ( + patch("kleinanzeigen_bot.published_ads.fetch_published_ads", new_callable = AsyncMock) as mock_fetch, + patch("kleinanzeigen_bot.delete_flow.delete_ad", new_callable = AsyncMock, return_value = DeleteResult(deleted = False, attempted = False)), + patch.object(test_bot, "web_sleep", new_callable = AsyncMock), + ): + await delete_flow.delete_ads( + web = test_bot, root_url = test_bot.root_url, + after_delete = test_bot.config.deleting.after_delete, + delete_old_ads_by_title = True, + ad_cfgs = [(ad_file, ad_cfg, ad_cfg_orig)], + ) + + mock_fetch.assert_not_awaited() + + @pytest.mark.asyncio + async def test_delete_ads_skips_title_deletes_but_continues_id_deletes_when_strict_fetch_fails( + self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any], tmp_path:Path, + ) -> None: + """Incomplete strict title matching should not block exact-ID deletes in the same batch.""" + test_bot.config.deleting.after_delete = "NONE" + title_ad_file, title_ad_cfg, title_ad_cfg_orig = self._make_ad(minimal_ad_config, tmp_path) + title_ad_cfg.id = None + title_ad_cfg_orig.pop("id", None) + id_ad_file, id_ad_cfg, id_ad_cfg_orig = self._make_ad(minimal_ad_config | {"title": "Exact ID Title"}, tmp_path) + + fetch_error = PublishedAdsFetchIncompleteError("page 2 timed out") + mock_delete = AsyncMock(return_value = DeleteResult(deleted = True, attempted = True)) + + with ( + patch("kleinanzeigen_bot.published_ads.fetch_published_ads", new_callable = AsyncMock, side_effect = fetch_error) as mock_fetch, + patch("kleinanzeigen_bot.delete_flow.delete_ad", new = mock_delete), + patch.object(test_bot, "web_sleep", new_callable = AsyncMock), + ): + await delete_flow.delete_ads( + web = test_bot, root_url = test_bot.root_url, + after_delete = test_bot.config.deleting.after_delete, + delete_old_ads_by_title = True, + ad_cfgs = [(title_ad_file, title_ad_cfg, title_ad_cfg_orig), (id_ad_file, id_ad_cfg, id_ad_cfg_orig)], + ) + + mock_fetch.assert_awaited_once_with(test_bot, test_bot.root_url, strict = True) + mock_delete.assert_awaited_once_with( + test_bot, test_bot.root_url, id_ad_cfg, [], delete_old_ads_by_title = True, + ) + @pytest.mark.asyncio async def test_cleanup_on_title_match_all_404_with_id_none( self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any], tmp_path:Path,